套接字：如何在接收/解析数据时向客户端发送数据而不“等待”它们

Question

我有一个套接字服务器，使用 boost::asio 用 C++ 编写，我正在向客户端发送数据。

服务器以块的形式发送数据，客户端在接收到每个块时对其进行解析。两者现在几乎都是单线程的。

我应该在服务器上使用什么设计来确保服务器尽可能快地写出数据，而不是等待客户端解析它？我想我需要在服务器上做一些异步操作。

我想也可以在客户端上进行更改以实现此目的，但理想情况下，无论客户端是如何编写的，服务器都不应该等待客户端。

我正在像这样向套接字写入数据：

size_t bytesWritten = m_Socket.Write( boost::asio::buffer(buffer, bufferSize));

更新：

我将尝试使用 Boost 的机制来异步写入套接字。见http://www.boost.org/doc/libs/1_36_0/doc/html/boost_asio/tutorial/tutdaytime3/src.html

例如

 boost::asio::async_write(socket_, boost::asio::buffer(message_),
        boost::bind(&tcp_connection::handle_write, shared_from_this(),
          boost::asio::placeholders::error,
          boost::asio::placeholders::bytes_transferred));

• 亚历克斯

Answer 1

如果您将套接字设置为非阻塞，那么如果它们会阻塞，写入应该会失败。然后，您可以根据自己的喜好对数据进行排队，并安排稍后再次尝试写入数据。我不知道如何在 boost socket API 中设置套接字选项，但这正是您要找的。​​p>

但这可能比它的价值更麻烦。您需要选择一个准备好写入的套接字，大概从同时打开的几个套接字中，将更多数据推入其中直到它已满，然后重复。我不知道 boost sockets API 是否有 select 的等价物，这样你就可以一次等待多个套接字，直到它们中的任何一个准备好写入。

服务器通常为每个客户端连接启动一个线程（或生成一个进程）的原因正是为了在等待 I/O 的同时继续为其他客户端提供服务，同时避免实现自己的队列。 “安排稍后再尝试”的最简单方法就是在专用线程中阻塞 I/O。

除非 boost 在其套接字 API 中做了一些不寻常的事情，否则你不能做的是要求操作系统或套接字库为你排队任意数量的数据而不会阻塞。可能有一个异步 API 会在数据写入时回调您。

Answer 2

您可以通过不通过 TCP 而是通过 UDP 传输数据来确保异步通信。 但是，如果您需要使用 TCP，请让客户端快速将数据存储起来，然后在不同的线程中或与 cron 作业异步处理。

Answer 3

当您将数据传递给套接字时，它不会等待接收者处理它。它甚至不等待数据传输。数据被放入由操作系统在后台处理的出站队列。写入函数返回排队等待传输的字节数，而不是实际传输的字节数。

Answer 4

继续 Stefan 帖子中的 cmets：

绝对可以在客户端或服务器端进行缓冲。但请务必考虑尼尔写的内容。如果我们只是开始盲目地缓冲数据，并且如果处理永远跟不上发送，那么我们的缓冲区将以我们可能不想要的方式增长。

现在我最近实现了一个简单的“NetworkPipe”，它旨在充当单个客户端/服务器、服务器/客户端之间的连接，外部用户不知道/关心管道是客户端还是服务器。我实现了类似于您所询问的缓冲情况，如何？好吧，这个类是线程化的，这是我能想出干净地缓冲数据的唯一方法。这是我遵循的基本过程，请注意我在管道上设置了最大尺寸：

1. 进程 1 启动管道，默认为服务器。现在内部线程等待客户端。
2. 进程 2 启动管道，已经是服务器，默认为客户端。
3. 我们现在已连接，首先要做的是交换最大缓冲区大小。
4. 进程 1 写入数据（它注意到另一端有一个空缓冲区 [参见 #3]）
5. 进程 2 的内部线程（现在正在等待套接字的 select()）看到数据已发送并读取它，并对其进行缓冲。进程 2 现在将新的缓冲大小发送回 P1。

所以这是一个非常简化的版本，但基本上通过线程化，我总是可以等待阻塞选择调用，一旦数据到达，我可以读取并缓冲它，我会发回新的缓冲大小。你可以做类似的事情，盲目地缓冲数据，它实际上相当简单，因为你不必交换缓冲区大小，但这可能是个坏主意。所以上面的例子允许外部用户在不阻塞线程的情况下读取/写入数据（除非另一端的缓冲区已满）。

Answer 5

我使用 boost::asio::async_write 方法实现了一个解决方案。

基本上：

• 每个客户端都有一个线程（我的线程正在执行 CPU 密集型工作）
• 随着每个线程积累一定量的数据，它使用 async_write 将其写入套接字，而不关心之前的写入是否已完成
• 代码小心管理套接字的生命周期和正在写出的数据缓冲区，因为 CPU 处理在所有数据写出之前完成

这对我很有效。这使服务器线程能够在其 CPU 工作完成后立即完成。

总体而言，客户端接收和解析其所有数据的时间减少了。同样，服务器花费在每个客户端上的时间（墙上的时钟）也会减少。

代码sn-p：

void SocketStream::Write(const char* data, unsigned int dataLength)
{
    // Make a copy of the data
    // we'll delete it when we get called back via HandleWrite
    char* dataCopy = new char[dataLength];
    memcpy( dataCopy,  data, dataLength );

    boost::asio::async_write
        (
        *m_pSocket,
        boost::asio::buffer(dataCopy, dataLength),
        boost::bind
            (
            &SocketStream::HandleWrite,                     // the address of the method to callback when the write is done
            shared_from_this(),                             // a pointer to this, using shared_from_this to keep us alive
            dataCopy,                                       // first parameter to the HandleWrite method
            boost::asio::placeholders::error,               // placeholder so that async_write can pass us values
            boost::asio::placeholders::bytes_transferred
            )
        );
}

void SocketStream::HandleWrite(const char* data, const boost::system::error_code& error, size_t bytes_transferred)
{
    // Deallocate the buffer now that its been written out
    delete data;

    if ( !error )
    {
        m_BytesWritten += bytes_transferred;
    }
    else
    {
        cout << "SocketStream::HandleWrite received error: " << error.message().c_str() << endl;
    }
}