c++ - 使用 Poco SocketReactor 处理多线程的好方法是什么？

Question

因此，我开始研究实现高容量客户端/服务器系统的替代方案，并且我目前正在研究 Poco 的 Reactor 框架，因为我现在将 Poco 用于我的大部分应用程序框架。

传入的数据包大小会非常小，所以我认为从客户端读取数据的角度来看它会工作得很好。但是基于客户端输入执行的操作将相对昂贵，并且可能需要卸载到另一个进程甚至另一个服务器。发送回客户端的响应有时会相当大。所以很明显，我不能在发生这种情况时阻塞 react 器线程。

所以我在想，如果我只是读取 react 器事件处理程序中的数据，然后将其传递给另一个处理数据的线程(池)，效果会更好。

我不太确定操作完成后将响应发送回客户端的过程。

我找不到太多关于使用该框架的最佳方式的信息。但我做了一些测试，看起来 react 堆会在套接字可写时反复触发 WritableNotification 事件。那么，最佳流程是将需要发送的数据在接收 WritableNotification 事件的对象中排队，并在每次接收到事件时发送小块？

更新:所以当我开始测试这个时，我惊恐地发现服务器 CPU 使用率在运行服务器应用程序的 CPU 上上升到 100% 与单个连接。但是经过一番挖掘，我发现我做错了什么。我发现我不需要在创建服务处理程序时注册 WritableNotification 事件，我只需要在有数据要发送时注册。然后，一旦发送了所有数据，我就应该注销事件处理程序。这样，当没有任何内容可发送时， react 器就不必一遍又一遍地调用事件处理程序。现在，即使有 100 个连接，我的 CPU 使用率也接近于 0。哇!

Answer 1

我写了一个从 SocketConnector 复制的 ServerConnector 类，但不调用套接字的连接，因为套接字已经连接，如果 react 堆是用 ServiceHandler 启动的，用于 TcpServerConnection 的 run() 函数中的通知，类 TcpServer 将启动一个新线程。所以，我得到了 reactor-partten 的多线程，但我不知道这是不是最好的方法。

ServerConnector 类

template <class ServiceHandler>
class ServerConnector
{
public:     
    explicit ServerConnector(StreamSocket& ss):
        _pReactor(0),
        _socket(ss)
        /// Creates a ServerConnector, using the given Socket.
    {
    }

    ServerConnector(StreamSocket& ss, SocketReactor& reactor):
        _pReactor(0),
        _socket(ss)
        /// Creates an acceptor, using the given ServerSocket.
        /// The ServerConnector registers itself with the given SocketReactor.
    {
        registerConnector(reactor);
        onConnect();
    }

    virtual ~ServerConnector()
        /// Destroys the ServerConnector.
    {
        unregisterConnector();
    }

//
// this part is same with SocketConnector
//

private:
    ServerConnector();
    ServerConnector(const ServerConnector&);
    ServerConnector& operator = (const ServerConnector&);

    StreamSocket&   _socket;
    SocketReactor* _pReactor;
};

Echo-Service是一个普通的ServiceHander

class EchoServiceHandler
{
public:
    EchoServiceHandler(StreamSocket& socket, SocketReactor& reactor):
        _socket(socket),
        _reactor(reactor)
    {
        _reactor.addEventHandler(_socket, Observer<EchoServiceHandler, ReadableNotification>(*this, &EchoServiceHandler::onReadable));
        _reactor.addEventHandler(_socket, Observer<EchoServiceHandler, ErrorNotification>(*this, &EchoServiceHandler::onError));
    }

    ~EchoServiceHandler()
    {
        _reactor.removeEventHandler(_socket, Observer<EchoServiceHandler, ErrorNotification>(*this, &EchoServiceHandler::onError));
        _reactor.removeEventHandler(_socket, Observer<EchoServiceHandler, ReadableNotification>(*this, &EchoServiceHandler::onReadable));
    }

    void onReadable(ReadableNotification* pNf)
    {
        pNf->release();
        char buffer[4096];
        try {
            int n = _socket.receiveBytes(buffer, sizeof(buffer));
            if (n > 0)
            {
                _socket.sendBytes(buffer, n);
            } else
                onError();
        } catch( ... ) {
            onError();
        }
    }

    void onError(ErrorNotification* pNf)
    {
        pNf->release();
        onError();
    }

    void onError()
    {
        _socket.shutdown();
        _socket.close();
        _reactor.stop();
        delete this;
    }

private:
    StreamSocket   _socket;
    SocketReactor& _reactor;
};

EchoReactorConnection 与类 TcpServer 一起将 react 器作为线程运行

class EchoReactorConnection: public TCPServerConnection
{
public:
    EchoReactorConnection(const StreamSocket& s): TCPServerConnection(s)
    {
    }

    void run()
    {
        StreamSocket& ss = socket();
        SocketReactor reactor;

        ServerConnector<EchoServiceHandler> sc(ss, reactor);
        reactor.run();
        std::cout << "exit EchoReactorConnection thread" << std::endl;
    }
};

cppunit 测试用例与 TCPServerTest::testMultiConnections 相同，但多线程使用 EchoReactorConnection。

void TCPServerTest::testMultithreadReactor()
{
    ServerSocket svs(0);
    TCPServerParams* pParams = new TCPServerParams;
    pParams->setMaxThreads(4);
    pParams->setMaxQueued(4);
    pParams->setThreadIdleTime(100);

    TCPServer srv(new TCPServerConnectionFactoryImpl<EchoReactorConnection>(), svs, pParams);
    srv.start();

    assert (srv.currentConnections() == 0);
    assert (srv.currentThreads() == 0);
    assert (srv.queuedConnections() == 0);
    assert (srv.totalConnections() == 0);

    //
    // same with TCPServerTest::testMultiConnections()
    //
    // ....
    ///
}