這篇文章給大家介紹怎樣進(jìn)行服務(wù)器程序的架構(gòu)分析，內(nèi)容非常詳細(xì)，感興趣的小伙伴們可以參考借鑒，希望對大家能有所幫助。

創(chuàng)新互聯(lián)公司主要從事成都網(wǎng)站制作、成都做網(wǎng)站、外貿(mào)營銷網(wǎng)站建設(shè)、網(wǎng)頁設(shè)計、企業(yè)做網(wǎng)站、公司建網(wǎng)站等業(yè)務(wù)。立足成都服務(wù)濱湖,10余年網(wǎng)站建設(shè)經(jīng)驗,價格優(yōu)惠、服務(wù)專業(yè),歡迎來電咨詢建站服務(wù):18980820575

下面將介紹我曾經(jīng)做過的一個項目的服務(wù)器架構(gòu)和服務(wù)器編程的一些重要細(xì)節(jié)。

程序運行環(huán)境

操作系統(tǒng)：Centos 7.0

編譯器：gcc/g++ 4.8.3、cmake 2.8.11

MySQL數(shù)據(jù)庫：5.5.47

項目代碼管理工具：Visual Studio 2013

一、程序結(jié)構(gòu)

該程序總共有 17 個線程，其中分為 9 個數(shù)據(jù)庫工作線程 D 和一個日志線程 L，6 個普通工作線程 W，一個主線程  M。(以下會用這些字母來代指這些線程)

(一)、數(shù)據(jù)庫工作線程的用途

9 個數(shù)據(jù)庫工作線程在線程啟動之初，與 mysql 建立連接，也就是說每個線程都與 mysql 保持一路連接，共 9 個數(shù)據(jù)庫連接。

每個數(shù)據(jù)庫工作線程同時存在兩個任務(wù)隊列，***個隊列 A 存放需要執(zhí)行數(shù)據(jù)庫增刪查改操作的任務(wù) sqlTask，第二個隊列 B 存放 sqlTask  執(zhí)行完成后的結(jié)果。sqlTask 執(zhí)行完成后立即放入結(jié)果隊列中，因而結(jié)果隊列中任務(wù)也是一個個的需要執(zhí)行的任務(wù)。大致偽代碼如下：

void db_thread_func()  {       while (!m_bExit)       {           if (NULL != (pTask = m_sqlTask.Pop()))           {               //從m_sqlTask中取出的任務(wù)先執(zhí)行完成后，pTask將攜帶結(jié)果數(shù)據(jù)               pTask->Execute();                           //得到結(jié)果后，立刻將該任務(wù)放入結(jié)果任務(wù)隊列               m_resultTask.Push(pTask);               continue;           }            sleep(1000);       }//end while-loop   }

現(xiàn)在的問題來了：

任務(wù)隊列 A 中的任務(wù)從何而來，目前只有消費者，沒有生產(chǎn)者，那么生產(chǎn)者是誰?

任務(wù)隊列 B 中的任務(wù)將去何方，目前只有生產(chǎn)者沒有消費者。

這兩個問題先放一會兒，等到后面我再來回答。

(二)工作線程和主線程

在介紹主線程和工作線程具體做什么時，我們介紹下服務(wù)器編程中常常抽象出來的幾個概念(這里以 tcp 連接為例)：

1. TcpServer 即 Tcp 服務(wù)，服務(wù)器需要綁定ip地址和端口號，并在該端口號上偵聽客戶端的連接(往往由一個成員變量 TcpListener  來管理偵聽細(xì)節(jié))。所以一個 TcpServer 要做的就是這些工作。除此之外，每當(dāng)有新連接到來時，TcpServer  需要接收新連接，當(dāng)多個新連接存在時，TcpServer 需要有條不紊地管理這些連接：連接的建立、斷開等，即產(chǎn)生和管理下文中說的TcpConnection  對象。

2. 一個連接對應(yīng)一個 TcpConnection 對象，TcpConnection 對象管理著這個連接的一些信息：如連接狀態(tài)、本端和對端的 ip  地址和端口號等。

3. 數(shù)據(jù)通道對象 Channel，Channel 記錄了 socket 的句柄，因而是一個連接上執(zhí)行數(shù)據(jù)收發(fā)的真正執(zhí)行者，Channel 對象一般作為  TcpConnection 的成員變量。

4. TcpSession 對象，是將 Channel 收取的數(shù)據(jù)進(jìn)行解包，或者對準(zhǔn)備好的數(shù)據(jù)進(jìn)行裝包，并傳給 Channel 發(fā)送。

歸納起來：一個 TcpServer 依靠 TcpListener 對新連接的偵聽和處理，依靠TcpConnection  對象對連接上的數(shù)據(jù)進(jìn)行管理，TcpConnection 實際依靠 Channel 對數(shù)據(jù)進(jìn)行收發(fā)，依靠 TcpSession 對數(shù)據(jù)進(jìn)行裝包和解包。也就是說一個  TcpServer 存在一個 TcpListener，對應(yīng)多個 TcpConnection，有幾個TcpConnection  就有幾個TcpSession，同時也就有幾個 Channel。

以上說的 TcpServer、TcpListener、TcpConnection、Channel 和 TcpSession  是服務(wù)器框架的網(wǎng)絡(luò)層。一個好的網(wǎng)絡(luò)框架，應(yīng)該做到與業(yè)務(wù)代碼脫耦。即上層代碼只需要拿到數(shù)據(jù)，執(zhí)行業(yè)務(wù)邏輯，而不用關(guān)注數(shù)據(jù)的收發(fā)和網(wǎng)絡(luò)數(shù)據(jù)包的封包和解包以及網(wǎng)絡(luò)狀態(tài)的變化(比如網(wǎng)絡(luò)斷開與重連)。

拿數(shù)據(jù)的發(fā)送來說：

當(dāng)業(yè)務(wù)邏輯將數(shù)據(jù)交給 TcpSession，TcpSession 將數(shù)據(jù)裝好包后(裝***程后可以有一些加密或壓縮操作)，交給  TcpConnection::SendData()，而TcpConnection::SendData() 實際是調(diào)用  Channel::SendData()，因為 Channel 含有 socket 句柄，所以Channel::SendData()  真正調(diào)用send()/sendto()/write() 方法將數(shù)據(jù)發(fā)出去。

對于數(shù)據(jù)的接收，稍微有一點不同：

通過 select()/poll()/epoll() 等IO multiplex技術(shù)，確定好了哪些 TcpConnection 上有數(shù)據(jù)到來后，激活該  TcpConnection 的 Channel 對象去調(diào)用recv()/recvfrom()/read() 來收取數(shù)據(jù)。數(shù)據(jù)收到以后，將數(shù)據(jù)交由  TcpSession來處理，最終交給業(yè)務(wù)層。注意數(shù)據(jù)收取、解包乃至交給業(yè)務(wù)層是一定要分開的。我的意思是：***不要解包并交給業(yè)務(wù)層和數(shù)據(jù)收取的邏輯放在一起。因為數(shù)據(jù)收取是  IO 操作，而解包和交給業(yè)務(wù)層是邏輯計算操作。IO 操作一般比邏輯計算要慢。到底如何安排要根據(jù)服務(wù)器業(yè)務(wù)來取舍，也就是說你要想好你的服務(wù)器程序的性能瓶頸在網(wǎng)絡(luò)  IO 還是邏輯計算，即使是網(wǎng)絡(luò)  IO，也可以分為上行操作和下行操作，上行操作即客戶端發(fā)數(shù)據(jù)給服務(wù)器，下行即服務(wù)器發(fā)數(shù)據(jù)給客戶端。有時候數(shù)據(jù)上行少，下行大。(如游戲服務(wù)器，一個 npc  移動了位置，上行是該客戶端通知服務(wù)器自己***位置，而下行確是服務(wù)器要告訴在場的每個客戶端)。

工作線程的流程：

while (!m_bQuit)   {       epoll_or_select_func();        handle_io_events();        handle_other_things();   }

其中 epoll_or_select_func() 即是上文所說的通過 select()/poll()/epoll() 等 IO multiplex  技術(shù)，確定好了哪些 TcpConnection 上有數(shù)據(jù)到來。我的服務(wù)器代碼中一般只會監(jiān)測 socket 可讀事件，而不會監(jiān)測 socket  可寫事件。至于如何發(fā)數(shù)據(jù)，文章后面會介紹。所以對于可讀事件，以 epoll 為例，這里需要設(shè)置的標(biāo)識位是：

• EPOLLIN  普通可讀事件(當(dāng)連接正常時，產(chǎn)生這個事件，recv()/read()函數(shù)返回收到的字節(jié)數(shù);當(dāng)連接關(guān)閉，這兩個函數(shù)返回0，也就是說我們設(shè)置這個標(biāo)識已經(jīng)可以監(jiān)測到新來數(shù)據(jù)和對端關(guān)閉事件)

• EPOLLRDHUP 對端關(guān)閉事件(linux man  手冊上說這個事件可以監(jiān)測對端關(guān)閉，但我實際調(diào)試時發(fā)送即使對端關(guān)閉也沒觸發(fā)這個事件，仍然是EPOLLIN，只不過此時調(diào)用recv()/read()函數(shù)，返回值會為0，所以實際項目中是否可以通過設(shè)置這個標(biāo)識來監(jiān)測對端關(guān)閉，仍然待考證)

• EPOLLPRI 帶外數(shù)據(jù)

muduo 里面將 epoll_wait 的超時事件設(shè)置為 1 毫秒，我的另一個項目將 epoll_wait 超時時間設(shè)置為 10  毫秒。這兩個數(shù)值供大家參考。

這個項目中，工作線程和主線程都是上文代碼中的邏輯，主線程監(jiān)聽偵聽socket 上的可讀事件，也就是監(jiān)測是否有新連接來了。主線程和每個工作線程上都存在一個  epollfd。如果新連接來了，則在主線程的 handle_io_events() 中接受新連接。產(chǎn)生的新連接的socket句柄掛接到哪個線程的 epollfd  上呢?這里采取的做法是 round-robin 算法，即存在一個對象CWorkerThreadManager  記錄了各個工作線程上工作狀態(tài)。偽碼大致如下：

void attach_new_fd(int newsocketfd)   {       workerthread = get_next_worker_thread(next);       workerthread.attach_to_epollfd(newsocketfd);       ++next;       if (next > max_worker_thread_num)           next = 0;   }

即先從***個工作線程的 epollfd 開始掛接新來  socket，接著累加索引，這樣下次就是第二個工作線程了。如果所以超出工作線程數(shù)目，則從***個工作重新開始。這里解決了新連接 socket  “負(fù)載均衡”的問題。在實際代碼中還有個需要注意的細(xì)節(jié)就是：epoll_wait 的函數(shù)中的 struct epoll_event  數(shù)量開始到底要設(shè)置多少個才合理?存在的顧慮是，多了浪費，少了不夠用，我在曾經(jīng)一個項目中直接用的是 4096：

const int EPOLL_MAX_EVENTS = 4096;   const int dwSelectTimeout = 10000;   struct epoll_event events[EPOLL_MAX_EVENTS];   int nfds = epoll_wait(m_fdEpoll, events, EPOLL_MAX_EVENTS, dwSelectTimeout / 1000);

我在陳碩的 muduo 網(wǎng)絡(luò)庫中發(fā)現(xiàn)作者才用了一個比較好的思路，即動態(tài)擴(kuò)張數(shù)量：開始是 n個，當(dāng)發(fā)現(xiàn)有事件的 fd 數(shù)量已經(jīng)到達(dá) n 個后，將  struct epoll_event 數(shù)量調(diào)整成 2n 個，下次如果還不夠，則變成 4n 個，以此類推，作者巧妙地利用 stl::vector  在內(nèi)存中的連續(xù)性來實現(xiàn)了這種思路：

//初始化代碼   std::vector events_(16);    //線程循環(huán)里面的代碼   while (m_bExit)   {       int numEvents = ::epoll_wait(epollfd_, &*events_.begin(), static_cast(events_.size()), 1);       if (numEvents > 0)       {           if (static_cast(numEvents) == events_.size())           {               events_.resize(events_.size() * 2);           }       }   }

讀到這里，你可能覺得工作線程所做的工作也不過就是調(diào)用 handle_io_events()  來接收網(wǎng)絡(luò)數(shù)據(jù)，其實不然，工作線程也可以做程序業(yè)務(wù)邏輯上的一些工作。也就是在handle_other_things() 里面。那如何將這些工作加到  handle_other_things() 中去做呢?寫一個隊列，任務(wù)先放入隊列，再讓 handle_other_things()  從隊列中取出來做?我在該項目中也借鑒了muduo庫的做法。即 handle_other_things() 中調(diào)用一系列函數(shù)指針，偽碼如下：

void do_other_things()   {       somefunc();   } //m_functors是一個stl::vector,其中每一個元素為一個函數(shù)指針   void somefunc()   {       for (size_t i = 0; i < m_functors.size(); ++i)       {           m_functors[i]();       }        m_functors.clear();   }

//m_functors是一個stl::vector,其中每一個元素為一個函數(shù)指針 void somefunc() { for (size_t i =  0; i < m_functors.size(); ++i) { m_functors[i](); } m_functors.clear(); }

當(dāng)任務(wù)產(chǎn)生時，只要我們將執(zhí)行任務(wù)的函數(shù) push_back 到 m_functors 這個 stl::vector  對象中即可。但是問題來了，如果是其他線程產(chǎn)生的任務(wù)，兩個線程同時操作 m_functors，必然要加鎖，這也會影響效率。muduo 是這樣做的：

void add_task(const Functor& cb)   {       std::unique_lock lock(mutex_);       m_functors.push_back(cb);      }    void do_task()   {       std::vector functors;       {           std::unique_lock lock(mutex_);           functors.swap(m_functors);       }        for (size_t i = 0; i < functors.size(); ++i)       {           functors[i]();       }   }

看到?jīng)]有，利用一個棧變量 functors 將 m_functors 中的任務(wù)函數(shù)指針倒換(swap)過來了，這樣大大減小了對 m_functors  操作時的加鎖粒度。前后變化：變化前，相當(dāng)于原來 A 給 B 多少東西，B 消耗多少，A 給的時候，B 不能消耗;B  消耗的時候A不能給?，F(xiàn)在變成A將東西放到籃子里面去，B 從籃子里面拿，B 如果拿去一部分后，只有消耗完了才會來拿，或者 A 通知 B 去籃子里面拿，而 B  忙碌時，A 是不會通知 B 來拿，這個時候 A 只管將東西放在籃子里面就可以了。

bool bBusy = false;   void add_task(const Functor& cb)   {       std::unique_lock lock(mutex_);       m_functors_.push_back(cb);        //B不忙碌時只管往籃子里面加，不要通知B       if (!bBusy)       {           wakeup_to_do_task();       }   }    void do_task()   {       bBusy = true;       std::vector functors;       {           std::unique_lock lock(mutex_);           functors.swap(pendingFunctors_);       }        for (size_t i = 0; i < functors.size(); ++i)       {           functors[i]();       }        bBusy = false;   }

看，多巧妙的做法!

因為每個工作線程都存在一個 m_functors，現(xiàn)在問題來了，如何將產(chǎn)生的任務(wù)均衡地分配給每個工作線程。這個做法類似上文中如何將新連接的 socket  句柄掛載到工作線程的 epollfd 上，也是 round-robin 算法。上文已經(jīng)描述，此處不再贅述。

還有種情況，就是希望任務(wù)產(chǎn)生時，工作線程能夠立馬執(zhí)行這些任務(wù)，而不是等 epoll_wait  超時返回之后。這個時候的做法，就是使用一些技巧喚醒epoll_wait，Linux 系統(tǒng)可以使用socketpair 或 timerevent、eventfd  等技巧。

問題 1 的答案是：業(yè)務(wù)層產(chǎn)生任務(wù)可能會交給數(shù)據(jù)庫任務(wù)隊列A，這里的業(yè)務(wù)層代碼可能就是工作線程中 do_other_things()  函數(shù)執(zhí)行體中的調(diào)用。至于交給這個 9 個數(shù)據(jù)庫線程的哪一個的任務(wù)隊列，同樣采用了 round-robin 算法。所以就存在一個對象  CDbThreadManager來管理這九個數(shù)據(jù)庫線程。下面的偽碼是向數(shù)據(jù)庫工作線程中加入任務(wù)：

bool CDbThreadManager::AddTask(IMysqlTask* poTask )   {       if (m_index >= m_dwThreadsCount)       {           m_index = 0;       }        return m_aoMysqlThreads[m_index++].AddTask(poTask);   }

同理問題 2 中的消費者也可能就是 do_other_things() 函數(shù)執(zhí)行體中的調(diào)用。

現(xiàn)在來說問題 3，業(yè)務(wù)層的數(shù)據(jù)產(chǎn)生后，經(jīng)過 TcpSession 裝包后，需要發(fā)送的話，產(chǎn)生任務(wù)丟給工作線程的  do_other_things()，然后在相關(guān)的 Channel 里面發(fā)送，因為沒有監(jiān)測該 socket 上的可寫事件，所以該數(shù)據(jù)可能調(diào)用 send() 或者  write() 時會阻塞，沒關(guān)系，sleep() 一會兒，繼續(xù)發(fā)送，一直嘗試，到數(shù)據(jù)發(fā)出去。偽碼如下：

bool Channel::Send()   {       int offset = 0;       while (true)       {           int n = ::send(socketfd, buf + offset, length - offset);           if (n == -1)           {               if (errno == EWOULDBLOCK)               {                   ::sleep(100);                   continue;               }           }           //對方關(guān)閉了socket，這端建議也關(guān)閉           else if (n == 0)           {               close(socketfd);               return false;           }            offset += n;           if (offset >= length)               break;        }        return true;       }

最后，還有一個模塊日志線程沒有介紹，高性能的日志實現(xiàn)方案目前并不常見。

關(guān)于怎樣進(jìn)行服務(wù)器程序的架構(gòu)分析就分享到這里了，希望以上內(nèi)容可以對大家有一定的幫助，可以學(xué)到更多知識。如果覺得文章不錯，可以把它分享出去讓更多的人看到。