muduo网络库——实现Boost.Asio聊天服务器:完整代码+注释

Bertha 。 2022-05-04 09:27 255阅读 0赞

**聊天服务**

一个服务端进程可以同时服务多个客户端，客户端接受键盘输入，以回车为界把消息发送给服务端，服务端收到消息之后，依次发送给每个连接到它的客户端，原来发送消息的客户端进程也会收到这条消息。

**消息格式**

每条消息有一个4字节头部，以网络序存放字节序长度。比如两条消息“hello”和“chenshuo"：

![70][]

**打包的代码**

把string message打包为muduo::net::Buffer,并通过conn发送。

void send(muduo::net::TcpConnection* conn, const muduo::StringPiece& message)
    {
        muduo::net::Buffer buf;
        buf.append(message.data(), message.size());        //添加数据
        int32_t len  = static_cast<int32_t>(message.size());
        
        //网络字节序是大端存储
        int32_t be32 = muduo::net::sockets::hostToNetwork32(len);    //读取长度
        buf.prepend(&be32, sizeof(be32));                //添加长度
        conn->send(&buf);                //发送数据
    }

**分包的代码**

void onMessage(const muduo::net::TcpConnectionPtr& conn,
                    muduo::net::Buffer* buf,
                    muduo::Timestamp receiveTime)
    {
        while(buf->readableBytes() >= kHeaderLen)    //kHeaderLen = 4
        {
            //FIXME:use Buffer::peekInt32()
            cont void* data = buf->peek();            //读取消息长度
            int32_t be32 = *static_cast<const int32_t*>(data);    //SIGBUS
            const int32_t len = muduo::net::sockets::networkToHost32(be32);
            if(len > 65536 || len < 0)
            {
                LOG_ERROR << "Invalid length " << len;
                conn->shutdown();        //FIXME: disable reading
                break;
            }
            else if(buf->readableBytes() >= len+kHeaderLen)
            {
                buf->retrieve(kHeaderLen);         //回收数据
                muduo::string message(buf->peek(), len);    //设置消息内容
                messageCallback_(conn, message, receiveTime);    //回调用户代码
                buf->retrieve(len);                //回收所有数据
            }
            else
            {
                break;
            }
        }
    }

用while循环来反复读取数据，直到Buffer中的数据不够一条完整的消息。

**编解码器LengthHeaderCodec**

用来解决头部信息的编解码，当已连接套接字可读时，muduo的TcpConnction会读数据并存入input buffer中，然后回调用户的函数。通过LengthHeaderCodec这一层封装，让用户代码只关心“消息到达”而不是“数据到达”。

//muduo/examples/asio/chat/codec.h
    
    #ifndef __CODEC_H__
    #define __CODEC_H__
    #include<muduo/base/Logging.h>
    #include<muduo/net/Buffer.h>
    #include<muduo/net/Endian.h>
    #include<muduo/net/TcpConnection.h>
    
    class LengthHeaderCodec:boost::noncopyable 
    {
        public:
            typedef boost::function<void (const muduo::net::TcpConnectionPtr&,
                                          const muduo::string& message,
                                          muduo::Timestamp)> StringMessageCallback; //回调类型
    
                //构造函数
            explicit LengthHeaderCodec(const StringMessageCallback& cb)
                :messageCallback_(cb)
            {
            }
    
            //onMessage()和send()同前
    
            void onMessage(const muduo::net::TcpConnectionPtr& conn,
                muduo::net::Buffer* buf,
                muduo::Timestamp receiveTime)
            {
                while (buf->readableBytes() >= kHeaderLen) // kHeaderLen == 4
                {
                    // FIXME: use Buffer::peekInt32()
                    const void* data = buf->peek();//读取长度
                    int32_t be32 = *static_cast<const int32_t*>(data); // SIGBUS
                    const int32_t len = muduo::net::sockets::networkToHost32(be32);
                    if (len > 65536 || len < 0)
                    {
                        LOG_ERROR << "Invalid length " << len;
                        conn->shutdown();  // FIXME: disable reading
                        break;
                    }
                    else if (buf->readableBytes() >= len + kHeaderLen)
                    {
                        buf->retrieve(kHeaderLen);//回收数据
                        muduo::string message(buf->peek(), len);//设置消息内容
                        messageCallback_(conn, message, receiveTime);
                        buf->retrieve(len);//回收所有数据
                    }
                    else
                    {
                        break;
                    }
                }
    
            }
            void send(muduo::net::TcpConnection* conn,
                const muduo::StringPiece& message)
            {
                muduo::net::Buffer buf;
                buf.append(message.data(), message.size());//添加数据
                int32_t len = static_cast<int32_t>(message.size());
                //网络字节序是大端存储
                int32_t be32 = muduo::net::sockets::hostToNetwork32(len);//读取长度
                buf.prepend(&be32, sizeof be32);//添加长度
                conn->send(&buf);//发送数据
            }//
        private:
            StringMessageCallback messageCallback_;
            const static size_t kHeaderLen = sizeof(int32_t);
    };
    #endif

有几个关于muduo::net::buffer相关的内容：  
1.buffer预设了8个字节的保留字节，在Buffer内部通过readerIndex\_管理（初始值为8）。  
2.retrieve(int kBytes)意为回收，在内部实际是改变readIndex\_的值，**表示kBytes的数据已读**。  
3.hostToNetwork32函数底层封装了htobe32（这是大端小端存储转换的函数），**网络字节序是大端存储**。

### 服务端的实现 ###

在muduo中设置多线程的方式很简单server.setThreadNum(threadNum) 。但muduo一直强调的one loop per thread（其实就是reactor）的思路还是要了解。  
回忆一下传统的单线程reactor过程：  
1.服务器socket->bind->listen->poll/select管理监听套接字，并把用一个fd数组保存监听套接字。  
2.连接到达，套接字可读，poll/select返回，将已连接套接字添加到fd数组，继续poll/select等待  
3.消息到达，套接字可读，相关处理。

放到多线程里面，如果某个线程作为base thread，该线程有一个main Reactor负责accept连接，然后把已连接套接字挂在某个sub Reactor中（I/O Thread），至于怎么选择，以达到每个工作线程的“负载均衡”，muduo采用round-robin(轮询调度)的方式。

//muduo/examples/asio/chat/server_thread.cc
    
    #include "codec.h"
    #include <sys/types.h>
    #include <muduo/base/Logging.h>
    #include <muduo/base/Mutex.h>
    #include <muduo/net/EventLoop.h>
    #include <muduo/net/TcpServer.h>
    
    #include <boost/bind.hpp>
    
    #include <set>
    #include <stdio.h>
    
    using namespace muduo;
    using namespace muduo::net;
    
    class ChatServer : boost::noncopyable
    {
     public:
    
    //首先,在构造函数里注册回调：
      ChatServer(EventLoop* loop,
                 const InetAddress& listenAddr)
      : server_(loop, listenAddr, "ChatServer"),
        codec_(boost::bind(&ChatServer::onStringMessage, this, _1, _2, _3))    //向codec_注册onStringMessage()
      {
        server_.setConnectionCallback(
            boost::bind(&ChatServer::onConnection, this, _1));
        server_.setMessageCallback(
            boost::bind(&LengthHeaderCodec::onMessage, &codec_, _1, _2, _3));
      }
    
      void setThreadNum(int numThreads)
      {
        server_.setThreadNum(numThreads);
      }
    
      void start()
      {
        server_.start();
      }
    
     private:
    
    //以下是处理连接的建立和断开的代码，它把新建的连接加入到connections_容器中，把已断开的连接从容器中删除。
      void onConnection(const TcpConnectionPtr& conn)
      {
        LOG_INFO << conn->localAddress().toIpPort() << " -> "
            << conn->peerAddress().toIpPort() << " is "
            << (conn->connected() ? "UP" : "DOWN");
    
        MutexLockGuard lock(mutex_);
        if (conn->connected())
        {
          connections_.insert(conn);    //添加
        }
        else
        {
          connections_.erase(conn);    //删除
        }
      }
    
    //以下是服务端处理消息的代码
      void onStringMessage(const TcpConnectionPtr&,const string& message,Timestamp)
      {
        MutexLockGuard lock(mutex_);
    //遍历整个connections_容器
        for (ConnectionList::iterator it = connections_.begin(); it != connections_.end(); ++it)
        {
          codec_.send(get_pointer(*it), message);    //把消息发送给各个客户连接
        }
      }
    
    //数据成员：
      typedef std::set<TcpConnectionPtr> ConnectionList;
      TcpServer server_;
      LengthHeaderCodec codec_;
      MutexLock mutex_;
      ConnectionList connections_;
    };
    
    int main(int argc, char* argv[])
    {
      LOG_INFO << "pid = " << getpid();
      if (argc > 1)
      {
        EventLoop loop;
        uint16_t port = static_cast<uint16_t>(atoi(argv[1]));
        InetAddress serverAddr(port);        //初始化sockaddr_in
        ChatServer server(&loop, serverAddr);    //初始化聊天服务器
        if (argc > 2)
        {
          server.setThreadNum(atoi(argv[2]));    //设置多线程
        }
        server.start();
        loop.loop();
      }
      else
      {
        printf("Usage: %s port [thread_num]\n", argv[0]);
      }
    }

**注意：**

在构造函数里注册回调函数时，以往是直接把本class的onMessage()注册给server\_,这里我们把LengthHeaderCodec::onMessage()注册给server\_,然后向codec\_注册了ChatServer::onStringMessage(),等于说让codec\_负责解析消息，然后把完整的消息回调给ChatServer。这是一个”**简单的间接层**“。如图

![70 1][]

### 客户端的实现 ###

两个线程：main()函数所在线程用来从标准输入读入发送的消息，另外一个线程用EventLoopThread处理网络I/O。

//muduo/examples/asio/chat/client.cc
    
    #include "codec.h"
    
    #include <muduo/base/Logging.h>
    #include <muduo/base/Mutex.h>
    #include <muduo/net/EventLoopThread.h>
    #include <muduo/net/TcpClient.h>
    
    #include <boost/bind.hpp>
    #include <boost/noncopyable.hpp>
    
    #include <iostream>
    #include <stdio.h>
    #include <unistd.h>
    
    using namespace muduo;
    using namespace muduo::net;
    
    class ChatClient : boost::noncopyable
    {
     public:
    //首先在构造函数里注册回调，并使用跟前面一样的LengthHeaderCodec作为中间层，负责打包、分包。
      ChatClient(EventLoop* loop, const InetAddress& serverAddr)
        : client_(loop, serverAddr, "ChatClient"),
          codec_(boost::bind(&ChatClient::onStringMessage, this, _1, _2, _3)) //向codec_注册ChatClient::onStringMessage()  
      {
        client_.setConnectionCallback(
            boost::bind(&ChatClient::onConnection, this, _1));
        client_.setMessageCallback(
            boost::bind(&LengthHeaderCodec::onMessage, &codec_, _1, _2, _3));
        client_.enableRetry();
      }
    
      void connect()
      {
        client_.connect();
      }
    
      void disconnect()
      {
        client_.disconnect();
      }
    
    //write()会由main线程调用
      void write(const StringPiece& message)
      {
        MutexLockGuard lock(mutex_);    //加锁保护shared_ptr
        if (connection_)
        {
          codec_.send(get_pointer(connection_), message);
        }
      }
    
     private:
    //onConnection()由EventLoop线程调用
      void onConnection(const TcpConnectionPtr& conn)
      {
        LOG_INFO << conn->localAddress().toIpPort() << " -> "
                 << conn->peerAddress().toIpPort() << " is "
                 << (conn->connected() ? "UP" : "DOWN");
    
        MutexLockGuard lock(mutex_);    //加锁保护shared_ptr
        if (conn->connected())
        {
          connection_ = conn;
        }
        else
        {
          connection_.reset();
        }
      }
    
    //.......由EventLoop线程调用,把消息打印到屏幕
      void onStringMessage(const TcpConnectionPtr&,
                           const string& message,
                           Timestamp)
      {
        printf("<<< %s\n", message.c_str());
      }
    
    //数据成员
      TcpClient client_;
      LengthHeaderCodec codec_;
      MutexLock mutex_;
      TcpConnectionPtr connection_ GUARDED_BY(mutex_);
    };
    
    int main(int argc, char* argv[])
    {
      LOG_INFO << "pid = " << getpid();
      if (argc > 2)
      {
        EventLoopThread loopThread;
        uint16_t port = static_cast<uint16_t>(atoi(argv[2]));
        InetAddress serverAddr(argv[1], port);    //初始化sockaddr_in
    
        ChatClient client(loopThread.startLoop(), serverAddr);    //初始化聊天客户端
        client.connect();                    //发起连接
        std::string line;
        while (std::getline(std::cin, line))    //读取键盘输入
        {
          client.write(line);    //发送数据行
        }
        client.disconnect();    //断开连接
        CurrentThread::sleepUsec(1000*1000);  // wait for disconnect, see ace/logging/client.cc
      }
      else
      {
        printf("Usage: %s host_ip port\n", argv[0]);
      }
    }

### **简单测试** ###

(可执行文件在muduo/build/release/bin目录中)

打开三个窗口，在第一个窗口运行：

./asio_chat_server 3000

第二个窗口运行：

./asio_chat_client 127.0.0.1 3000

第三个窗口运行：

./asio_chat_client 127.0.0.1 3000

这样就有两个客户端进程参与聊天，第二个窗口里输入有些字符并回车，字符会出现在本窗口和第三个窗口中。

### 多线程测试 ###

同样打开三个窗口，第一个窗口运行：

./asio_chat_server_threaded 3000 2    //设置线程数为2

第二、第三个窗口与之前相同。

服务器端：

![70 2][]

客户端发送和接收消息：

![70 3][]

![70 4][]

服务线程有三个（**1 base reactor+2 sub reactors**）

![70 5][]

客户线程有两个：

（每个客户进程都有两个线程）

client0:

![70 6][]

client1:

![70 7][]

[70]: /images/20220504/cebe490e970846a0a991d9728583efd0.png
[70 1]: /images/20220504/bc9312e598704c3aba3fcfd619bdeaa7.png
[70 2]: /images/20220504/bdbc94b33bce45e38e276d6fc420a5d1.png
[70 3]: /images/20220504/e0754a671ae641588ee1ed96250dc473.png
[70 4]: /images/20220504/bd9bdc81037247418e0016bdaf042342.png
[70 5]: /images/20220504/d9bb7b81b05545bba2a26d1a2ae56678.png
[70 6]: /images/20220504/ded5af9ad1744275832792d1023e2cba.png
[70 7]: /images/20220504/0ded76cc65ad49b385936e8f1263837e.png