muduo实现聊天服务器

- 日理万妓 2022-07-16 09:11 220阅读 0赞

# muduo实现聊天服务器 #

muduo实现一个聊天室服务器，客户发送的消息将广播到连入的所有客户（包括自己）。

## **1.消息编码** ##

消息的字节流定义成这种形式 `0xXX 0xXX 0xXX 0xXX XXXXXX`，前面四个字节表示消息的长度，后面是消息实体，在服务器收到消息后，需要将该消息解码，把前面四个字节的长度信息剥离，把消息发送出去。  
muduo作者选择自己编写一个编解码器`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)
            {
            }
            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（这是大端小端存储转换的函数），网络字节序是大端存储。

--------------------

## **2.服务器的实现** ##

书中提供了三个思路：  
1.单线程Reactor  
2.多线程Reactor（one loop per thread）使用mutex保护共享数据。  
3.多线程+shared\_ptr

在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))
      {
        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:
      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_);
        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);
        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]);
      }
    }

--------------------

## **3.客户端实现** ##

两个线程，一个线程用来从标准输入读入发送的消息，另外一个线程用Reactor处理网络I/O，这里用两个线程的原因是因为作者没有把标准输入输出加入到Reactor的想法，在UNP的单线程Reactor中有管理0,1,2（标准输入、标准输出、标准错误）监听读入键盘数据的示例。

//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>
    
    using namespace muduo;
    using namespace muduo::net;
    
    class ChatClient : boost::noncopyable
    {
     public:
      ChatClient(EventLoop* loop, const InetAddress& serverAddr)
        : client_(loop, serverAddr, "ChatClient"),
          codec_(boost::bind(&ChatClient::onStringMessage, this, _1, _2, _3))
      {
        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();
      }
    
      void write(const StringPiece& message)
      {
        MutexLockGuard lock(mutex_);
        if (connection_)
        {
          codec_.send(get_pointer(connection_), message);
        }
      }
    
     private:
      void onConnection(const TcpConnectionPtr& conn)
      {
        LOG_INFO << conn->localAddress().toIpPort() << " -> "
                 << conn->peerAddress().toIpPort() << " is "
                 << (conn->connected() ? "UP" : "DOWN");
    
        MutexLockGuard lock(mutex_);
        if (conn->connected())
        {
          connection_ = conn;
        }
        else
        {
          connection_.reset();
        }
      }
    
      void onStringMessage(const TcpConnectionPtr&,
                           const string& message,
                           Timestamp)
      {
        printf("<<< %s\n", message.c_str());
      }
    
      TcpClient client_;
      LengthHeaderCodec codec_;
      MutexLock mutex_;
      TcpConnectionPtr connection_;
    };
    
    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);
    
        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]);
      }
    }

--------------------

## **4.测试** ##

这里主要测试和验证的内容：  
1.消息广播  
2.查看客户端的两个线程（main+Reactor）。  
3.服务器有threadNum+1个线程（base Reactor+sub Reactors）

测试情况如下：

**服务器绑定端口号设置线程数为2：**  
![这里写图片描述][20161012230557784]  
**客户端发送和接收消息：**  
![这里写图片描述][20161012230626816]  
**服务器线程有3个（1 base reactor+2 sub reactors）**  
![这里写图片描述][20161012230802599]  
**客户线程有2个（1 main + 1 reactor）**  
![这里写图片描述][20161012230911804]

--------------------

## **5.参考** ##

1.UNP  
2.Linux多线程服务端编程  
3.[查看进程中的多线程][Link 1]

[20161012230557784]: /images/20220716/30cbcdc5e9574f8397e832b54ce92366.png
[20161012230626816]: /images/20220716/041690fd42cd4bedafba42f8c117ee1d.png
[20161012230802599]: /images/20220716/1d37e50914a54fc3b86c70aed233766a.png
[20161012230911804]: /images/20220716/c9efbeed8bb443d592b69d7213e7d13e.png
[Link 1]: http://www.codeceo.com/article/linux-find-thread.html