完整的使用線程池的多線程C/S Socket類

完整的使用線程池的多線程C/S Socket類

翻譯水平有限，不明之處請閱讀原文。

原文：http://www.codeproject.com/Articles/33352/Full-Multi-thread-Client-Server-Socket-Class-with

 

使用線程池實現的完整的 Client/Server Socket通訊類，很容易使用，也很容易被集成到C++應用程序中。也適用於Linux/Unix。

在“Best C++/MFC article of February 2009”賽中獲獎。

代碼下載請到原文地址。

當做客戶端運行時，在命令行中輸入：SocketServer.exe /client

簡介

最近我在Code Project中更新了一篇文章：ServerSocket。雖然其基類（CSocketHandle）非常穩定、易用，但必須承認，一些最初的用以保持通訊接口完整性的設計對新的開發來說開始成了一個問題。

本文將解決這個問題，我提出了一個新的、改進版本的通訊類，並向你展示如何使用線程池提高你的網絡應用的性能。

描述

首先，我假設你已經熟悉了socket編程，並在你的工作領域有了幾年的經驗。如果不是這樣，我強烈向你推薦幾個鏈接，它們可能會對你有幫助。它們是“all fired up and ready to go”，請閱讀。我將盡力說明如何使用這個新類，以增強你的系統性能。 

同步Socket

默認地，socket工作在阻塞模式，這就意味着你需要一個專門的線程來read/wait數據，同時需要另一個線程來write/send數據。現在使用新的模板類就變得容易多了。

典型地，一個客戶端只需要一個線程，所以不會有什么問題，但如果你在開發服務器組件，並需要可靠的通訊或者是連接客戶端的P2P，你遲早會發現自己需要多線程來處理請求。

SocketClientImpl

第一個模板SocketClientImpl封裝了客戶端的socket通訊，它可以使用TCP（SOCK_STREAM）或UDP（SOCK_DGRAM）通訊。這里的好消息是，它操控一個通訊循環，並用一種高效的方式來報告數據和幾個重要的事件。這些對你來說，任務就真正變得直截了當。

template <typename T, size_t tBufferSize = 2048>
class SocketClientImpl
{
    typedef SocketClientImpl<T, tBufferSize> thisClass;
public:
    SocketClientImpl()
    : _pInterface(0)
    , _thread(0)
    {
    }

    void SetInterface(T* pInterface)
    {
        ::InterlockedExchangePointer(reinterpret_cast<void**>(&_pInterface), pInterface);
    }

    bool IsOpen() const;
    bool CreateSocket(LPCTSTR pszHost, LPCTSTR pszServiceName,
            int nFamily, int nType, UINT uOptions = 0);
    bool ConnectTo(LPCTSTR pszHostName, LPCTSTR pszRemote,
            LPCTSTR pszServiceName, int nFamily, int nType);
    void Close();
    DWORD Read(LPBYTE lpBuffer, DWORD dwSize,
        LPSOCKADDR lpAddrIn = NULL, DWORD dwTimeout = INFINITE);
    DWORD Write(const LPBYTE lpBuffer, DWORD dwCount,
        const LPSOCKADDR lpAddrIn = NULL, DWORD dwTimeout = INFINITE);
    bool StartClient(LPCTSTR pszHost, LPCTSTR pszRemote,
        LPCTSTR pszServiceName, int nFamily, int nType);
    void Run();
    void Terminate(DWORD dwTimeout = 5000L);

    static bool IsConnectionDropped(DWORD dwError);

protected:
    static DWORD WINAPI SocketClientProc(thisClass* _this);
    T*              _pInterface;
    HANDLE          _thread;
    CSocketHandle   _socket;
};

客戶端接口報告以下事件：

class ISocketClientHandler
{
public:
    virtual void OnThreadBegin(CSocketHandle* ) {}
    virtual void OnThreadExit(CSocketHandle* ) {}
    virtual void OnDataReceived(CSocketHandle* , const BYTE* ,
                DWORD , const SockAddrIn& ) {}
    virtual void OnConnectionDropped(CSocketHandle* ) {}
    virtual void OnConnectionError(CSocketHandle* , DWORD ) {}
};

 

函數	描述
OnThreadBegin	線程啟動時調用
OnThreadExit	線程要退出時調用
OnDataReceived	有新數據到來時調用
OnConnectionDropped	偵測到一個錯誤時調用，錯誤是由網絡連接斷開或socket被關閉引起的。
OnConnectionError	偵測到一個錯誤時調用

這個接口實際上是很隨意的，即是可選的，你的程序可以這樣實現：

 

class CMyDialog : public CDialog
{
    typedef SocketClientImpl<CMyDialog> CSocketClient; // CMyDialog handles events!
public:
    CMyDialog(CWnd* pParent = NULL);   // standard constructor
    virtual CMyDialog ();

    // ...
    void OnThreadBegin(CSocketHandle* ) {}
    void OnThreadExit(CSocketHandle* ) {}
    void OnDataReceived(CSocketHandle* , const BYTE* , DWORD , const SockAddrIn& ) {}
    void OnConnectionDropped(CSocketHandle* ) {}
    void OnConnectionError(CSocketHandle* , DWORD ) {}

protected:
    CSocketClient m_SocketClient;
};

 

SocketServerImpl

第二個模板SocketServerImpl處理服務端所有的通訊任務。在UDP模式下，它與客戶端的行為非常相似。在TCP模式下，在一個單獨的線程池內，它代理管理每一個網絡連接。線程池模板是在MSDN下通過Kenny Kerr (^)發布的修改版本，你應該可以在你的工程中重用它，而不會有任何問題。其好處是它可以用以從一個線程池中調用一個類成員。回調可以是以下簽名：

void ThreadFunc();
void ThreadFunc(ULONG_PTR);

要記住，為使用QueueUserWorkItem，你需要Windows 2000 或更高版本，除非你在使用Windows CE，否則一般是不會有問題的。我所知道的，已經沒有人使用Windows 95/98了。:-)

class ThreadPool
{
    static const int MAX_THREADS = 50;
    template <typename T>
    struct ThreadParam
    {
        void (T::* _function)(); T* _pobject;
        ThreadParam(void (T::* function)(), T * pobject)
        : _function(function), _pobject(pobject) { }
    };
public:
    template <typename T>
    static bool QueueWorkItem(void (T::*function)(),
                                  T * pobject, ULONG nFlags = WT_EXECUTEDEFAULT)
    {
        std::auto_ptr< ThreadParam<T> > p(new ThreadParam<T>(function, pobject) );
        WT_SET_MAX_THREADPOOL_THREADS(nFlags, MAX_THREADS);
        bool result = false;
        if (::QueueUserWorkItem(WorkerThreadProc<T>,
                                p.get(),
                                nFlags))
        {
            p.release();
            result = true;
        }
        return result;
    }

private:
    template <typename T>
    static DWORD WINAPI WorkerThreadProc(LPVOID pvParam)
    {
        std::auto_ptr< ThreadParam<T> > p(static_cast< ThreadParam<T>* >(pvParam));
        try {
            (p->_pobject->*p->_function)();
        }
        catch(...) {}
        return 0;
    }

    ThreadPool();
};

 

template <typename T, size_t tBufferSize = 2048>
class SocketServerImpl
{
    typedef SocketServerImpl<T, tBufferSize> thisClass;
public:
    SocketServerImpl()
    : _pInterface(0)
    , _thread(0)
    {
    }

    void SetInterface(T* pInterface)
    {
        ::InterlockedExchangePointer(reinterpret_cast<void**>
                    (&_pInterface), pInterface);
    }

    bool IsOpen() const
    bool CreateSocket(LPCTSTR pszHost,
        LPCTSTR pszServiceName, int nFamily, int nType, UINT uOptions);
    void Close();
    DWORD Read(LPBYTE lpBuffer, DWORD dwSize,
        LPSOCKADDR lpAddrIn, DWORD dwTimeout);
    DWORD Write(const LPBYTE lpBuffer, DWORD dwCount,
        const LPSOCKADDR lpAddrIn, DWORD dwTimeout);
    bool Lock()
    {
        return _critSection.Lock();
    }

    bool Unlock()
    {
        return _critSection.Unlock();
    }

    bool CloseConnection(SOCKET sock);
    void CloseAllConnections();
    bool StartServer(LPCTSTR pszHost,
        LPCTSTR pszServiceName, int nFamily, int nType, UINT uOptions);
    void Run();
    void Terminate(DWORD dwTimeout);
    void OnConnection(ULONG_PTR s);

    static bool IsConnectionDropped(DWORD dwError);

protected:
    static DWORD WINAPI SocketServerProc(thisClass* _this);
    T*              _pInterface;
    HANDLE          _thread;
    ThreadSection   _critSection;
    CSocketHandle   _socket;
    SocketList      _sockets;
};

服務器接口報告以下事件：

class ISocketServerHandler
{
public:
    virtual void OnThreadBegin(CSocketHandle* ) {}
    virtual void OnThreadExit(CSocketHandle* )  {}
    virtual void OnThreadLoopEnter(CSocketHandle* ) {}
    virtual void OnThreadLoopLeave(CSocketHandle* ) {}
    virtual void OnAddConnection(CSocketHandle* , SOCKET ) {}
    virtual void OnRemoveConnection(CSocketHandle* , SOCKET ) {}
    virtual void OnDataReceived
    (CSocketHandle* , const BYTE* , DWORD , const SockAddrIn& ) {}
    virtual void OnConnectionFailure(CSocketHandle*, SOCKET) {}
    virtual void OnConnectionDropped(CSocketHandle* ) {}
    virtual void OnConnectionError(CSocketHandle* , DWORD ) {}
};

這個接口也是可選的，但我希望你使用它，以使設計更干凈。

異步Socket

Windows支持異步socket，你可以使用通訊類CSocketHandle來實現。你需要為你的工程定義SOCKHANDLE_USE_OVERLAPPED，異步通訊是非阻塞模式，因此它允許你在單個線程中處理多個請求，你也可能需要多個“讀/寫”緩沖區來等待I/O。異步socket是個大課題，可能需要單獨一篇文章來討論它，但我希望你考慮一下當前支持的這個設計，函數CSocketHandle::ReadEx 和 CSocketHandle::WriteEx 讓你可以使用這種模式，最新的模板ASocketServerImpl展示了在異步讀取模式下如何使用SocketHandle類。主要的好處是，在TCP模式下，用一個線程處理所有的網絡連接。

結論

本文中，我向你介紹了CSocketHandle類的新改進，我希望這個新接口讓你工作更容易。當然，我一直廣納諫言，你可以自由地就此議題提出問題和建議。

參考文獻

• Windows Sockets
• Berkeley Sockets (Wikipedia)