Muduo 多線程模型對比


  本文主要對比Muduo多線程模型方案8 和方案9 。

  方案8:reactor + thread pool ,有一個線程來充當reactor 接受連接分發事件,將要處理的事件分配給thread pool中的線程,由thread pool 來完成事件處理。實例代碼見:examples/sudoku/server_threadpool.cc

  這里截取關鍵部分代碼進行說明。

class  SudokuServer
{
  public  :
  SudokuServer(EventLoop* loop,  const  InetAddress& listenAddr,  int  numThreads)
    : loop_(loop),
      server_(loop, listenAddr,  "SudokuServer" ),
      numThreads_(numThreads),
      startTime_(Timestamp::now())
  {
    server_.setConnectionCallback(
        boost::bind(&SudokuServer::onConnection,  this , _1));
    server_.setMessageCallback(
        boost::bind(&SudokuServer::onMessage,  this , _1, _2, _3));
  }
 
   void  start()
  {
    LOG_INFO <<  "starting "  << numThreads_ <<  " threads." ;
    threadPool_.start(numThreads_);  // 注意這里,threadPool 的類型是: ThreadPool,且位置在start 里面
    server_.start();
  }
 
  private  :
   void  onConnection( const  TcpConnectionPtr& conn)
  {
    LOG_TRACE << conn->peerAddress().toIpPort() <<  " -> "
        << conn->localAddress().toIpPort() <<  " is "
        << (conn->connected() ?  "UP"  :  "DOWN" );
  }
 
   void  onMessage( const  TcpConnectionPtr& conn, Buffer* buf, Timestamp)
  {
...
         if  (!processRequest(conn, request)) // 封裝計算任務執行方法
        {
          conn->send(  "Bad Request!\r\n" );
          conn->shutdown();
           break ;
        }
      }
...
    }
  }
 
   bool  processRequest( const  TcpConnectionPtr& conn,  const  string& request)
  {
...
 
     if  (puzzle.size() == implicit_cast<size_t>(kCells))
    {
      threadPool_.run(boost::bind(&solve, conn, puzzle, id)); // 將計算任務轉移到 threadPool 線程
    }
     else
    {
      goodRequest =  false ;
    }
     return  goodRequest;
  }
 
   static  void  solve( const  TcpConnectionPtr& conn,
                     const  string& puzzle,
                     const  string& id)
  {
    LOG_DEBUG << conn->name();
    string result = solveSudoku(puzzle); // solveSudou 是一個pure function, 是可重入的 
     if  (id.empty())
    {
      conn->send(result+  "\r\n" );
    }
     else
    {
      conn->send(id+  ":" +result+  "\r\n" );
    }
  }
 
  EventLoop* loop_;
  TcpServer server_;
  ThreadPool threadPool_;  // 注意類型,方案8, reactor + threadpool
   int  numThreads_;
  Timestamp startTime_;
};
 
void  ThreadPool::start(  int  numThreads)  // 創建 thread pool,具體thread 調度這里暫時不分析
{
  assert(threads_.empty());
  running_ =  true ;
  threads_.reserve(numThreads);
   for  ( int  i = 0; i < numThreads; ++i)
  {
     char  id[32];
    snprintf(id,  sizeof  id,  "%d" , i);
    threads_.push_back(  new  muduo::Thread(
          boost::bind(&ThreadPool::runInThread,  this ), name_+id));
    threads_[i].start();
  }
}
 
方案9:main-reactor + subreactors, one loop per thread, 有一個主線程來扮演main-reactor 專門語句 accept 連接,其它線程負責讀寫文件描述符(socket)
 
class  SudokuServer
{
  public  :
  SudokuServer(EventLoop* loop,  const  InetAddress& listenAddr,  int  numThreads)
    : loop_(loop),
      server_(loop, listenAddr,  "SudokuServer" ),
      numThreads_(numThreads),
      startTime_(Timestamp::now())
  {
    server_.setConnectionCallback(
        boost::bind(&SudokuServer::onConnection,  this , _1));
    server_.setMessageCallback(
        boost::bind(&SudokuServer::onMessage,  this , _1, _2, _3));
    server_.setThreadNum(numThreads);  // 設置 EventLoopThreadPool里面的thread數量
  }
 
   void  start()
  {
    LOG_INFO <<  "starting "  << numThreads_ <<  " threads." ;
    server_.start();
  }
 
  private  :
   void  onConnection( const  TcpConnectionPtr& conn)
  {
    LOG_TRACE << conn->peerAddress().toIpPort() <<  " -> "
        << conn->localAddress().toIpPort() <<  " is "
        << (conn->connected() ?  "UP"  :  "DOWN" );
  }
 
   void  onMessage( const  TcpConnectionPtr& conn, Buffer* buf, Timestamp)
  {
...
         if  (!processRequest(conn, request))  //准備計算
        {
          conn->send(  "Bad Request!\r\n" );
          conn->shutdown();
           break ;
        }
...
    }
  }
 
   bool  processRequest( const  TcpConnectionPtr& conn,  const  string& request)
  {
...
     if  (puzzle.size() == implicit_cast<size_t>(kCells))
    {
      LOG_DEBUG << conn->name();
      string result = solveSudoku(puzzle);  // 計算在當前線程完成
       if  (id.empty())
      {
        conn->send(result+  "\r\n" );
      }
...
  }
  // 注意這里沒有類型為ThreadPool的 threadPool_成員,整個類使用Muduo默認線程模型的EventLoopThreadPool,TcpServer 聚合了EventLoopThreadPool
  EventLoop* loop_;
  TcpServer server_;
   int  numThreads_;
  Timestamp startTime_;
};
 
 
void  TcpServer::setThreadNum(  int  numThreads)
{
  assert(0 <= numThreads);
  threadPool_->setThreadNum(numThreads);  // 設置了 EventLoopThreadPool 里面的線程個數,為后面的threadPool_->start()服務
}
 
void  TcpServer::start()
{
   if  (!started_)
  {
    started_ =  true ;
    threadPool_->start(threadInitCallback_);  // TcpServer 中的 threadPool 類型是 EventLoopThreadPool
  }
 
   if  (!acceptor_->listenning())
  {
    loop_->runInLoop(
        boost::bind(&Acceptor::listen, get_pointer(acceptor_)));
  }
}
 
void  EventLoopThreadPool::start(  const  ThreadInitCallback& cb) // 開啟線程的方式是使用EventLoopThread,這個類將EventLoop 和 Thread 封裝在一起實現 one loop per thread
{
  assert(!started_);
  baseLoop_->assertInLoopThread();
 
  started_ =  true ;
 
   for  ( int  i = 0; i < numThreads_; ++i)
  {
    EventLoopThread* t =  new  EventLoopThread(cb); // 設置線程的 callback
    threads_.push_back(t); 
    loops_.push_back(t->startLoop()); // 保存loop方便管理和分配任務,任務分配其實是通過EventLoop::runInLoop() 來進行的
  }
   if  (numThreads_ == 0 && cb)
  {
    cb(baseLoop_);
  }
}
 
  總結一下,這里所謂的Reactor就是持有Poller的結構(稍微有點狹隘,這里先就這樣理解),Poller負責事件監聽和分發。持有EventLoop的結構就持有Poller。
  對於方案8只有一個類持有EventLoop,也就是只創建了一個EventLoop,這個Loop就是reactor,其它的Thread 是通過ThreadPool來實現的,因此只有reactor所在的線程來完成I/O,其它線程用於完成計算任務,所以說這個模型適合於計算密集型而不是I/O密集型。
  對於方案9,存在多個Reactor,其中main reactor 持有Acceptor,專門用於監聽三個半事件中的連接建立,消息到達和連接斷開以及消息發送事件都讓sub reactor來完成。由於main reactor 只關心連接建立事件,能夠適應高並發的IO請求,多個subreactor的存在也能兼顧I/O與計算,因此被認為是一個比較好的方案。
   后面還會深入學習Muduo網絡庫相關的內容,包括Reactor結構的簡化,線程池的實現,現代C++的編寫方式,使用C++11進行重寫等。現在看來C++11 thread library 提供的接口基本可以替換 posix thread library,雖然底層也許是通過posix thread實現的,畢竟Linux內核針對NPTL進行過修改。C++11 提供了 thread_local 來描述 線程局部存儲,但是沒有pthread_key_create() 提供 destructor那樣的功能,或者遇到需要使用TLS的地方轉過來使用posix 提供的接口。
 
Muduo 多線程 線程池 reactor


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM