muduo源码分析之多线程TcpServer

淡淡的烟草味﹌ 2022-07-14 14:50 214阅读 0赞

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

## **（一）EventLoopThread类** ##

在[muduo源码分析之EventLoop::runInLoop()函数][muduo_EventLoop_runInLoop] 因为要测试跨线程`runInLoop()`调用，因此，引入了`EventLoopThread`类，这里回顾以下。  
考虑两个函数：  
1.`EventLoopThread::threadFunc()`是线程函数，用来在线程中创建`EventLoop`对象，并获取该对象指针。  
2.`EventLoopThread::startLoop()`用来从上述函数返回`EventLoop`对象指针。  
**这两个函数通过条件变量保证线程启动后，对象地址赋值成功后才返回其指针。**  
具体来看源码：

///EventLoopThread::threadFunc()
    
    void EventLoopThread::threadFunc()
    {
      EventLoop loop;//创建EventLoop对象
      {
        MutexLockGuard lock(mutex_);
        loop_ = &loop;//获得EventLoop对象的指针
        //获得loop对象地址后再唤醒阻塞的线程
        cond_.notify();//唤醒条件变量
      }
      loop.loop();//线程中loop
      //assert(exiting_);
    }

///EventLoopThread::startLoop()
    
    EventLoop* EventLoopThread::startLoop()
    {
      assert(!thread_.started());
      thread_.start();//线程启动
      //之后启动线程，当前线程由于条件变量阻塞
      {
        MutexLockGuard lock(mutex_);
        while (loop_ == NULL)//条件变量
        {
          cond_.wait();//blocked
        }
      }
      //被唤醒后返回指针以保证指针的有效性
      return loop_;
    }

因此，通过`EventLoopThread`可以完成两件事情：  
1.启动线程并在线程中进行loop。  
2.获取线程中EventLoop对象的地址。

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

## **（二）EventLoopThreadPoll类** ##

顾名思义，初始化阶段根据用户输入创建特定数目的线程和`EventLoop`对象，其实现很简单：

void EventLoopThreadPoll::start()
    {
        baseLoop_->assertInLoopThread();                                                
    
        started_ = true;                                                                
    
        for (int i = 0; i < numThreads_; ++i)                                           
        {                                           
        //创建numThreads个线程，并且分别返回其EventLoop对象指针                     
          EventLoopThread* t = new EventLoopThread;                                     
          threads_.push_back(t);//EventLoopThread线程对象                                              
    
          ///vector<EventLoop*>loops_;
          loops_.push_back(t->startLoop());//EventLoop对象指针                          
        } 
    }

`EventLoopThreadPoll`可以创建n个线程/Loop 对于一个`TcpServer`来说，我们如何进行分发？  
对于muduo多线程工作的方式应该是这样的：

> 1.首先，作者一直强调的一个概念`one loop per thread`，在一个线程中只有一个事件循环EventLoop。  
> 2.对于muduo来说，如果设置`threadNums_=n`，那么服务器程序将创建n+1个线程，n个I/O线程，1个base 线程。  
> 3.在 base 线程中有一个main Reactor负责处理连接，也就是监听套接字描述符的Reactor  
> 4.当每有一个连接，accept返回的已连接套接字描述符将添加到I/O线程的Reactor中。  
> 5.对于4.中添加的方式，muduo没有考虑负载均衡，而是使用Round-Robin的方式分发描述符。

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

## **（三）单线程的TcpServer** ##

回顾单线程TcpServer的工作方式。

> 1.在main函数中（主线程）创建`EventLoop`对象，并用来初始化`TcpServer`。  
> 2.TcpServer中初始化监听描述符（用Acceptor封装）并添加到loop中。  
> 3.当有连接，监听套接字可读，`accept(2)`返回已连接套接字描述符，使用这个描述符构造`TcpConnection`  
> 4.同时也使用`TcpServer`中的`EventLoop`对象去初始化`TcpConnection`。  
> 5.这样，监听套接字描述符和已连接套接字描述符在一个`EventLoop`中，也就是在UNP中的poll/epoll管理多个套接字描述符，经典的单线程Reactor方式。

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

## **（四）多线程的TcpServer** ##

在上文中，我们已经通过`EventLoopThreadPoll`创建了多个线程，并在每个线程中创建一个`EventLoop`对象，并且在`EventLoopThreadPoll`对象所在的线程拿到这些线程和对象的指针。

现在我们就要用线程池中的`EventLoop`对象（I/O Loop）来初始化 `TcpConnection`，这个过程就将已连接套接字描述符分发到不同线程的事件循环中去了。

/// EventLoopThreadPoll::getNextLoop()
    
    //TcpServer每新建一个TcpConnection就会调用该函数
      //来取得一个EventLoop对象
    EventLoop* EventLoopThreadPool::getNextLoop()
      {
        baseLoop_->assertInLoopThread();
        EventLoop* loop = baseLoop_;
    
        if (!loops_.empty())
        {
          // round-robin
          // 简单的循环
          loops_中保存着所有的I/O Loop
          loop = loops_[next_];//
          ++next_;
          if (static_cast<size_t>(next_) >= loops_.size())
          {
            next_ = 0;
          }
        }
        return loop;//返回一个EventLoop对象
      }

因此，在`TcpServer`中，每当有新连接的时候，我们不用原来的`base loop`去初始化`TcpConnection`，而是通过调用`EventLoopThreadPoll::getNextLoop()`拿到一个`I/O loop`完成初始化。

/// TcpConnection::newConnection()
    
    void TcpServer::newConnection(int sockfd, const InetAddress& peerAddr)
    {
      loop_->assertInLoopThread();
      char buf[32];
      snprintf(buf, sizeof buf, "#%d", nextConnId_);
      ++nextConnId_;
      std::string connName = name_ + buf;
    
      LOG_INFO << "TcpServer::newConnection [" << name_
               << "] - new connection [" << connName
               << "] from " << peerAddr.toHostPort();
      InetAddress localAddr(sockets::getLocalAddr(sockfd));
      // FIXME poll with zero timeout to double confirm the new connection
      EventLoop* ioLoop = threadPool_->getNextLoop();//拿到一个I/O Loop
      TcpConnectionPtr conn(
          new TcpConnection(ioLoop, connName, sockfd, localAddr, peerAddr));
      connections_[connName] = conn;
      conn->setConnectionCallback(connectionCallback_);
      conn->setMessageCallback(messageCallback_);
      conn->setWriteCompleteCallback(writeCompleteCallback_);
      conn->setCloseCallback(
          boost::bind(&TcpServer::removeConnection, this, _1)); // FIXME: unsafe
    
          //由于不在本线程进行Loop，需要跨线程调用
      ioLoop->runInLoop(boost::bind(&TcpConnection::connectEstablished, conn));
    }

以上，muduo网络库中多线程TcpServer的大致框架和工作模式。

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

## **（五）参考** ##

1.Linux多线程服务端编程 使用muduoC++ 网络库  
2.[http://blog.csdn.net/zhangxiao93/article/details/53366739?locationNum=1&fps=1][muduo_EventLoop_runInLoop]  
3.[http://blog.csdn.net/zhangxiao93/article/details/53390180][http_blog.csdn.net_zhangxiao93_article_details_53390180]

[muduo_EventLoop_runInLoop]: http://blog.csdn.net/zhangxiao93/article/details/53366739?locationNum=1&fps=1
[http_blog.csdn.net_zhangxiao93_article_details_53390180]: http://blog.csdn.net/zhangxiao93/article/details/53390180