睿豐德科技 專注RFID識別技術和條碼識別技術與管理軟件的集成項目。質量追溯系統、MES系統、金蝶與條碼系統對接、用友與條碼系統對接

Poco::TCPServer框架解析

POCO C++ Libraries提供一套 C++ 的類庫用以開發基于網絡的可移植的應用程序，功能涉及線程、文件、流，網絡協議包括：HTTP、FTP、SMTP 等，還提供 XML 的解析和 SQL 數據庫的訪問接口。不僅給我的工作帶來極大的便利，而且設計巧妙，代碼易讀，注釋豐富，也是非常好的學習材料，我個人非常喜歡。POCO的創始人在這個開源項目的基礎上做了一些收費產品，也成立了自己的公司，"I am in the lucky position to work for my own company"，真是讓人羨慕啊。

POCO C++ Libraries


基于POCO的產品


Poco主頁：http://pocoproject.org/
Poco文檔：http://pocoproject.org/docs/
創始人主頁：http://obiltschnig.com/
公司主頁：http://www.appinf.com/

我主要用過Net，Data，XML部分，Net里對socket的封裝類，實現TCP，HTTP，SMTP協議的框架，Data里對MySQL接口封裝，XML里對DOM標準的實現。我目前主要是做C++網絡編程，曾經嘗試接觸ACE庫，但覺得太復雜難理解，而且也沒有條件用于項目，后來發現了Poco，不僅簡單易用，而且也包含《C++ Networking Programming》中提到的各種模式和框架，更難得的是文檔注釋豐富，看起源碼來相當舒服。因此想寫一些筆記，一方面借此加深對網絡編程的理解，另一方面也希望和大家多多交流。

第一篇討論Poco::TCPServer框架，先看看Poco::TCPServer有哪些類

class Net_API TCPServer: public Poco::Runnable
class Net_API TCPServerConnection: public Poco::Runnable
class Net_API TCPServerConnectionFactory
class Net_API TCPServerDispatcher: public Poco::Runnable
class Net_API TCPServerParams: public Poco::RefCountedObject

Runnable是一個抽象類，要求其子類實現一個run()函數，run()一般作為線程的入口函數.
先看看run函數是如何被調用的，以TCPServer為例。
 

void TCPServer::start()
{
    poco_assert (_stopped);

    _stopped = false;
    _thread.start(*this);//這里傳入本類對象地址
}

再看看Thread類的start函數
 

class Foundation_API Thread: private ThreadImpl

 

void Thread::start(Runnable& target)
{
    startImpl(target);
}

這里的ThreadImpl::startImpl是POSIX實現版本
 

void ThreadImpl::startImpl(Runnable& target)
{
      
    _pData->pRunnableTarget = ⌖
    if (pthread_create(&_pData->thread, &attributes, runnableEntry, this))
    {
        _pData->pRunnableTarget = 0;
        throw SystemException("cannot start thread");
    }
      

pRunnableTarget就指向Runnable了

void* ThreadImpl::runnableEntry(void* pThread)
{
      
    ThreadImpl* pThreadImpl = reinterpret_cast<ThreadImpl*>(pThread);
    AutoPtr<ThreadData> pData = pThreadImpl->_pData;
    try
    {
        pData->pRunnableTarget->run();//在這里
    }
    catch (Exception& exc)
      

知道run()是如何被調用了，再看看TcpServer::run()的實現

 1void TCPServer::run()
 2{
 3    while (!_stopped)
 4    {
 5        Poco::Timespan timeout(250000);
 6        if (_socket.poll(timeout, Socket::SELECT_READ))
 7        {
 8            try
 9            {
10                StreamSocket ss = _socket.acceptConnection();
11                // enabe nodelay per default: OSX really needs that
12                ss.setNoDelay(true);
13                _pDispatcher->enqueue(ss);
14            }
15            catch (Poco::Exception& exc)
16            

在第6行調用標準select函數，準備好讀后，第10行調用標準accept建立連接，然后把StreamSocke對象放入TCPServerDispatcher對象內的隊列中，可見TCPServer只是建立連接，之后的工作就都交給TCPServerDispatcher了。

在講TCPServerDispatcher之前，先需要說明的是第10行TCPServer的_socket成員變量類型是ServerSocket，ServerSocket在一個構造函數中調用了bind和listen，具體如下

ServerSocket::ServerSocket(Poco::UInt16 port, int backlog): Socket(new ServerSocketImpl)
{
    IPAddress wildcardAddr;
    SocketAddress address(wildcardAddr, port);
    impl()->bind(address, true);
    impl()->listen(backlog);
}

要注意，盡管ServerSocket類提供了無參構造函數，但使用無參構造函數創建的對象，在TCPServer對象調用start()之前，必須先bind和listen。
"  /// Before start() is called, the ServerSocket passed to
  /// TCPServer must have been bound and put into listening state."

繼續看TCPServerDispatcher，沿著TcpServer::run()第13行，我們看它的equeue()

 1void TCPServerDispatcher::enqueue(const StreamSocket& socket)
 2{
 3    FastMutex::ScopedLock lock(_mutex);
 4
 5    if (_queue.size() < _pParams->getMaxQueued())
 6    {
 7        _queue.enqueueNotification(new TCPConnectionNotification(socket));
 8        if (!_queue.hasIdleThreads() && _currentThreads < _pParams->getMaxThreads())
 9        {
10            try
11            {
12                static const std::string threadName("TCPServerConnection");
13                _threadPool.start(*this, threadName);
14                ++_currentThreads;
15            }
16            catch (Poco::Exception&)
17            

第7行enqueueNotification把一個TCPConnectionNotification入隊，第13行把this交給ThreadPool啟動一個線程，那么這個線程就要運行TCPServerDispatcher的run函數了，看TCPServerDispatcher::run()

 1void TCPServerDispatcher::run()
 2{
 3    AutoPtr<TCPServerDispatcher> guard(this, true); // ensure object stays alive
 4
 5    int idleTime = (int) _pParams->getThreadIdleTime().totalMilliseconds();
 6
 7    for (;;)
 8    {
 9        AutoPtr<Notification> pNf = _queue.waitDequeueNotification(idleTime);
10        if (pNf)
11        {
12            TCPConnectionNotification* pCNf = dynamic_cast<TCPConnectionNotification*>(pNf.get());
13            if (pCNf)
14            {
15                std::auto_ptr<TCPServerConnection> pConnection(_pConnectionFactory->createConnection(pCNf->socket()));
16                poco_check_ptr(pConnection.get());
17                beginConnection();
18                pConnection->start();
19                endConnection();
20            }
21        }
                 

第9行waitDequeueNotification一個Notification，第12行把這個通知類型轉換成TCPConnectionNotification，聯系之前的enqueueNotification，大概能才到是什么意思。第15行又出現個TCPServerConnection。好吧，看來搞清楚TCPServerDispatcher還是要先看下TCPServerConnection，還有TCPConnectionNotification。

盡管TCPServerConnection繼承了Runable，但沒有實現run()，它的start()如下

void TCPServerConnection::start()
{
    try
    {
        run();
    }

用戶需要繼承TCPServerConnection實現run函數，看下源碼中的說明，要完成對這個socket的處理，因為run函數返回時，連接就自動關閉了。

class Net_API TCPServerConnection: public Poco::Runnable
    /// The abstract base class for TCP server connections
    /// created by TCPServer.
    ///
    /// Derived classes must override the run() method
    /// (inherited from Runnable). Furthermore, a
    /// TCPServerConnectionFactory must be provided for the subclass.
    ///
    /// The run() method must perform the complete handling
    /// of the client connection. As soon as the run() method
    /// returns, the server connection object is destroyed and
    /// the connection is automatically closed.
    ///
    /// A new TCPServerConnection object will be created for
    /// each new client connection that is accepted by
    /// TCPServer.

連接在哪被關閉的呢，注釋說隨著TCPServerConnection對象銷毀而關閉。具體就是，TCPServerConnection類有個StreamSocket成員，StreamSocket繼承自Socket，Socket類又包含了個SocketImpl成員，所以就有~TCPServerConnection()->~StreamSocket()->~Socket()->~SocketImpl()，最后~SocketImpl()調用close()關閉了連接。

那么TCPServerConnection對象何時被創建何時被銷毀呢，這下又回到TCPServerDispatcher::run()來了，看TCPServerDispatcher::run()的第15行，創建了TCPServerConnection對象，第18行調用TCPServerConnection::start()進而調用TCPServerConnection::run()，第20行塊結束，TCPServerConnection對象隨著智能指針銷毀而被銷毀。

還剩TCPConnectionNotification或者說Notification要搞清楚了，但是要對Poco的Notifactions模塊源碼進行分析的話，本文的篇幅也就太長了，就從文檔來大致看看吧

Notifications
Facilities for type-safe sending and delivery of notification objects within a single thread or from one thread to another, also well suited for the implementation of notification mechanisms. A notification queue class for distributing tasks to worker threads, simplifying the implementation of multithreaded servers.

簡單的說Notifications模塊可用于線程間傳遞消息，簡化多線程服務器的實現。具體到TCPServer，就是把已連接的socket，放到NotificationQueue中，并從TheadPool出來一個線程，線程從NotificationQueue取到這個socket，從而用TCPServerConnection::run()里的邏輯對socket進行處理。

至此TCPServer基本分析完畢，還有TCPServerConnectionFactory和TCPServerParams，分別用于產生TCPServerConnection和設置參數，就不細說了。

縱觀Poco::TCPServer，套一下《Unix Network Programming》里講的服務器模型，屬于“在程序啟動階段創建一個線程池之后讓主線程調用accept，并把每個客戶連接傳遞給池中某個可用線程”。Poco1.3.6版里用select作為IO multiplexing。1.3.7版正在嘗試epoll(windows平臺依然是select)，但還未release，跟蹤svn來看也沒有使用edge-triggered模式。套用《The C10K problem》里講的服務器模型，屬于"Serve many clients with each thread, and use nonblocking I/O and level-triggered readiness notification"。總結起來就是：non-blocking IO + IO multiplexing(level-triggered) + thread pool。

Poco::TCPServer也許并不能算一個性能很高的TCP服務器，但我非常喜歡它的設計和編碼風格。順便提一下對底層socket的封裝，由socket類派生的各種子類，ServerSocket在構造函數中進行bind和listen，StreamSocket在構造函數進行connect，都是非常貼心的設計。
 
下一篇打算分析Poco的Reator框架。

RFID管理系統集成商 RFID中間件 條碼系統中間層 物聯網軟件集成