Leader/Follower多線程網絡模型介紹

以前分享過《輕量級 web server Tornado代碼分析》，介紹了目前咱們採用nginx + tornado的方式搭建升級、配管、數據中心等各種服務組建客戶端迭代體系。最近注意到，淘寶目前公開了其網絡服務器源代碼Tengine。根據官方介 紹，Tengine是由淘寶網發起的Web服務器項目。它在Nginx的基礎上，針對大訪問量網站的需求，添加了不少高級功能和特性。Tengine的性 能和穩定性已經在大型的網站如淘寶網，天貓商城等獲得了很好的檢驗。它的最終目標是打造一個高效、穩定、安全、易用的Web平臺。它們都採用了多線程非阻 塞模式，並使用了LF模型。我最近整理了一下LF的相關資料，和你們分享一下。對淘寶開源的Tengine有興趣的同窗能夠到這裏checkout代碼研 究：http://code.taobao.org/svn/tengine/trunk

一、 引言

你們知道，多線程網絡服務最簡單的方式就是一個鏈接一個線程，這種模型當客戶端鏈接數快速增加是就會出現性能瓶頸。固然，這時候，咱們理所固然會考慮使用線程池，而任何池的使用，都會帶來一個管理和切換的問題。 在java 1.4中引入了NIO編程模型，它採用了Reactor模式，或者說觀察者模式，因爲它的讀寫操做都是無阻塞的，使得咱們可以只用一個線程處理全部的IO 事件，這種處理方式是同步的。爲了提升性能，當一個線程收到事件後，會考慮啓動一個新的線程去處理，而本身繼續等待下一個請求。這裏可能會有性能問題，就 是把工做交給別一個線程的時候的上下文切換，包括數據拷貝。今天向你們介紹一種Leader-Follower模型。

二、 基本思想

全部線程會有三種身份中的一種：leader和 follower，以及一個幹活中的狀態：proccesser。它的基本原則就是，永遠最多隻有一個leader。而全部follower都在等待成爲 leader。線程池啓動時會自動產生一個Leader負責等待網絡IO事件，當有一個事件產生時，Leader線程首先通知一個Follower線程將 其提拔爲新的Leader，而後本身就去幹活了，去處理這個網絡事件，處理完畢後加入Follower線程等待隊列，等待下次成爲Leader。這種方法 能夠加強CPU高速緩存類似性，及消除動態內存分配和線程間的數據交換。

三、 原理分析

顯然地，經過預先分配一個線程池，Leader/Follower設計避免了動態線程建立和銷燬的額外開銷。將線程放在一個自組織的池中，並且無需交換數據，這種方式將上下文切換、同步、數據移動和動態內存管理的開銷都降到了最低。

 

不過，這種模式在處理短暫的、原子的、反覆的和基於事件的動做上能夠取得明顯的性能提 升，好比接收和分發網絡事件或者向數據庫存儲大量數據記錄。事件處理程序所提供的服務越多，其體積也就越大，而處理一個請求所需的時間越長，池中的線程佔 用的資源也就越多，同時也須要更多的線程。相應的，應用程序中其它功能可用的資源也就越少，從而影響到應用程序的整體性能、吞吐量、可擴展性和可用性。

在大多數LEADER/FOLLOWERS設計中共享的事件源封裝在一個分配器組件 中。若是在一個設計中聯合使用了LEADER/FOLLOWERS和REACTOR事件處理基礎設施，由reactor組件進行分發。封裝事件源將事件分 離和分派機制與事件處理程序隔離開來。每一個線程有兩個方法：一個是join方法，使用這個方法能夠把新初始化的線程加入到池中。新加入的線程將本身的執行 掛起到線程池監聽者條件(monitor condition)上，並開始等待被提高爲新的Leader。在它變成一個Leader以後，它即可以訪問共享的事件源，等待執行下一個到來的事件。另 一個是promote_new_leader方法，當前的Leader線程使用這個方法能夠提高新的Leader，其作法是經過線程池監聽者條件通知休眠 的Follower。收到通知的Follower繼續執行(resume)線程池的join方法，訪問共享事件源，並等待下一個事件的到來。

四、 代碼演示

首先用一段簡單的代碼演示一下整個角色轉換的過程。因爲同一時刻只有一個leader，用一個互斥量就能夠解決了。每一個線程一直在作以下4個步驟循環：

 

public class WorkThread     {         public static Mutex mutex = new Mutex();         public void start()         {             while (true)             {                 // 等待成爲leader                 waitToLeader();                 // 用select或epoll等方式等待消息處理                 simulateReactor();                 // 產生下一個leader                 promoteNewLeader();                 // 處理消息                 simulateDojob();             }         }         private void simulateDojob()         { …         }         private void promoteNewLeader()         {             Console.WriteLine(Thread.CurrentThread.Name + ": Release leadership to others..");             mutex.ReleaseMutex();         }         private void simulateReactor()         {             …         }         private void waitToLeader()         {             Console.WriteLine(Thread.CurrentThread.Name + ": Waiting to be Leader..");             mutex.WaitOne();         }     }

詳細的代碼能夠參見附件。

五、 代碼分析

接下來咱們來看一下一個典型的開源代碼實現：spserver。抄段官網的話，spserver 是一個實現了半同步/半異步(Half-Sync/Half-Async)和領導者/追隨者(Leader/Follower) 模式的服務器框架，可以簡化 TCP server 的開發工做。spserver 使用 c++ 實現，目前實現瞭如下功能：

Ø  封裝了 TCP server 中接受鏈接的功能

Ø  使用非阻塞型I/O和事件驅動模型，由主線程負責處理全部 TCP 鏈接上的數據讀取和發送，所以鏈接數不受線程數的限制

Ø  主線程讀取到的數據放入隊列，由一個線程池處理實際的業務

Ø  一個 http 服務器框架，即嵌入式 web 服務器

Spserver的每一個版本都有必定的修改。早先版本V0.5尚未引入 Leader/Follower模式，在V0.8版本中已經有了sp_lfserver。在V0.9版本中將其改成了sp_iocplfserver，引 入了iocp完成端口的名字，但事實上以前版本已經使用了完成端口的技術。簡單地說，iocp就是事件io操做由操做系統完成，完成後才由線程接收處理事 件。先看一下代碼,server啓動之後開始監聽，並將線程池啓動起來。線程入口函數lfHandler一直在循環執行handleOneEvent：

 

int SP_LFServer :: run() {      int ret = 0;      int listenFD = -1;      ret = SP_IOUtils::tcpListen( mBindIP, mPort, &listenFD, 0 );      if( 0 == ret ) {          mThreadPool = new SP_ThreadPool( mMaxThreads );          for( int i = 0; i < mMaxThreads; i++ ) {               mThreadPool->dispatch( lfHandler, this );          }      }      return ret; } void SP_LFServer :: lfHandler( void * arg ) {      SP_LFServer * server = (SP_LFServer*)arg;      for( ; 0 == server->mIsShutdown; ) {          server->handleOneEvent();      } }

接下來看一下handleOneEvent的處理，和上面的演示程序同樣，先 mutexlock爭取leader權，而後去等待讀、寫事件，最後釋放leadership給其它人，本身執行讀完成事件處理函數 task->run()或寫事件的完成端口事件completionMessage，這個completionMessage會作一些清理工做，例 如delete msg：

 

void SP_LFServer :: handleOneEvent() {      SP_Task * task = NULL;      SP_Message * msg = NULL;      pthread_mutex_lock( &mMutex );      for( ; 0 == mIsShutdown && NULL == task && NULL == msg; ) {          if( mEventArg->getInputResultQueue()->getLength() > 0 ) {               task = (SP_Task*)mEventArg->getInputResultQueue()->pop();          } else if( mEventArg->getOutputResultQueue()->getLength() > 0 ) {               msg = (SP_Message*)mEventArg->getOutputResultQueue()->pop();          }          if( NULL == task && NULL == msg ) {               event_base_loop( mEventArg->getEventBase(), EVLOOP_ONCE );          }      }      pthread_mutex_unlock( &mMutex );      if( NULL != task ) task->run();      if( NULL != msg ) mCompletionHandler->completionMessage( msg ); }

六、 框架使用

和以前介紹的框架同樣，採用spserver構建server很是快捷,以下，只要把SP_TestHandler裏的幾個處理事件實現便可。

 

class SP_TestHandler : public SP_Handler { public:      SP_ TestHandler (){}      virtual ~SP_ TestHandler (){}      virtual int start( SP_Request * request, SP_Response * response ) {}      virtual int handle( SP_Request * request, SP_Response * response ) {}      virtual void error( SP_Response * response ) {}      virtual void timeout( SP_Response * response ) {}      virtual void close() {} };   class SP_TestHandlerFactory : public SP_HandlerFactory { public:      SP_ TestHandlerFactory () {}      virtual ~SP_ TestHandlerFactory () {}      virtual SP_Handler * create() const {          return new SP_TestHandler();      } };   int main( int argc, char * argv[] ) {      int port = 3333, maxThreads = 4, maxConnections = 20000;      int timeout = 120, reqQueueSize = 10000;      const char * serverType = "lf";      SP_IocpLFServer server( "", port, new SP_TestHandlerFactory() );      server.setTimeout( timeout );      server.setMaxThreads( maxThreads );      server.setMaxConnections( maxConnections );      server.runForever();      return 0; }

    Spserver的代碼能夠在這裏看到：http://spserver.googlecode.com/svn/trunk/spserver/。spserver同時實現了一個與leader/follower齊名的網絡編程模型：HAHS，翻譯爲半異步半同步模型。本文暫不做介紹。