带线程池的完整多线程客户端/服务器套接字类

• 下载 SocketHandle - 51.41 KB
• 下载 SocketServer - 260.48 KB
• 下载 SocketPort_Linux - 13.89 KB

要将应用程序作为客户端运行，请从命令提示符键入 SocketServer.exe /client。

引言

最近，我更新了我在 The Code Project 的早期文章之一，即 ServerSocket。虽然基础类 (CSocketHandle) 非常稳定且易于使用，但不得不承认，最初的一些设计决策是为了保持通信接口的完整性，而这对于新的开发来说正逐渐成为一个问题。

这就是本文的目标，我将展示这个通信类的新版本和改进版本，并展示如何利用线程池来提高网络解决方案的性能。

描述

首先，我假设您已经熟悉套接字编程，并且拥有多年的经验。如果不是这样，我强烈推荐您参考参考部分的一些链接，这些链接可能会为您指明方向。对于那些“已经准备好出发”的人，请继续阅读。我将尝试阐明如何使用新类来提高系统的性能。

同步套接字

默认情况下，套接字以阻塞模式运行，这意味着您需要一个专门的线程来读取/等待数据，而另一个线程将在另一端写入/发送数据。现在使用新的模板类使这一切变得更容易。

通常，客户端只需要一个线程，所以这没有问题，但是如果您正在开发服务器组件，需要可靠的通信或与客户端的点对点链接，迟早，您会发现需要多个线程来处理您的请求。

SocketClientImpl

第一个模板 SocketClientImpl 从客户端的角度封装了套接字通信。它可以用于使用 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	检测到错误时调用。该错误由连接丢失或套接字关闭引起。
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 模式下，它将每个连接的管理委托给一个单独的池化线程。池化线程模板是 Kenny Kerr 在 MSDN 上发布的一个修改版本 (^)。您应该能够毫无问题地在您的项目中重用它。好处是它可以用于从线程池中调用类成员。回调可以具有以下签名

    void ThreadFunc();
    void ThreadFunc(ULONG_PTR);

请记住，您需要 Windows 2000 或更高版本才能使用 QueueUserWorkItem。除非您针对的是 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 ) {}
};

此接口也是可选的，但我希望您会决定使用它，因为它使设计更清晰。

异步套接字

Windows 支持异步套接字。通信类 CSocketHandle 也使您能够访问它。您需要为您的项目定义 SOCKHANDLE_USE_OVERLAPPED。异步通信是非阻塞模式，因此允许您在单个线程中处理多个请求。您还可以提供多个读/写缓冲区来排队您的 I/O。异步套接字是一个大主题，可能值得单独写一篇文章，但我希望您会考虑当前支持的设计。CSocketHandle::ReadEx 和 CSocketHandle::WriteEx 函数让您可以访问此模式。最新的模板 ASocketServerImpl 展示了如何在异步读取模式下使用 SocketHandle 类。主要优点是，在 TCP 模式下，只使用一个线程来处理您的所有连接。

结论

在本文中，我将介绍 CSocketHandle 类的最新改进。我希望新接口能让您更轻松。当然，我随时欢迎您的建议，请随时在讨论区提出您的问题和建议。
尽情享用！

参考

• Windows 套接字
• Berkeley Sockets (Wikipedia)

历史

• 2009年12月2日 - 首次发布（作为独立文章发布）
• 2009年2月17日 - 更新了 Windows XP 的 ThreadPool 启动标志
• 2009年3月14日 - 修复了模板中的挂起问题（99% CPU）
• 2009年3月29日 - 异步模式服务器模板（+ 支持：WindowsCE、UNIX/Linux）
• 2009年4月5日 - 修复了服务器模板中的资源泄漏
• 2009年8月7日 - 修复了异步模式构建（使用 SOCKHANDLE_USE_OVERLAPPED）
• 2009年9月26日 - IPv6 支持（Windows 和 Linux）