我一直在构建一个用于多媒体消息传递的高吞吐量服务器应用程序，实现语言是C++。每个服务器都可以独立使用，也可以将许多服务器连接在一起以创建基于 DHT 的覆盖网络；服务器就像 Skype 中的 super 节点一样。

工作正在进行中。目前，服务器每秒可以处理大约 200,000 条消息(256 字节消息)，并且在我的机器(Intel i3 Mobile 2 GHz、Fedora Core 18(64 位)、4 GB RAM)上的最大吞吐量约为 256 MB/s长度为 4096 字节的消息。服务器有两个线程，一个线程用于处理所有 IO(基于 epoll，边缘触发)，另一个线程用于处理传入消息。覆盖管理还有另一个线程，但在当前讨论中无关紧要。

讨论中的两个线程使用两个循环缓冲区共享数据。线程号 1 使用一个循环缓冲区为线程号 2 排队新消息，而线程号 2 通过另一个循环缓冲区返回处理过的消息。服务器是完全无锁的。我没有使用任何同步原语，甚至没有使用原子操作。

循环缓冲区永远不会溢出，因为消息是池化的(在开始时预先分配)。事实上，所有重要/经常使用的数据结构都被合并以减少内存碎片并提高缓存效率，因此我们知道服务器启动时我们将要创建的最大消息数，因此我们可以预先计算最大缓冲区的大小，然后相应地初始化循环缓冲区。

现在我的问题是:线程 #1 一次将一条消息(实际上是指向消息对象的指针)的序列化消息排入队列，而线程 #2 从中取出消息队列中的 block (32/64/128 的 block )，并通过第二个循环缓冲区以 block 的形式返回处理后的消息。如果没有新消息，线程 #2 会一直忙着等待，因此让其中一个 CPU 核心一直处于忙碌状态。我怎样才能进一步改进设计？繁忙等待策略的替代方案是什么？我想优雅而高效地做到这一点。我考虑过使用信号量，但我担心这不是最佳解决方案，原因很简单，每次我在线程 #1 中排队消息时都必须调用“sem_post”，这可能会大大减少吞吐量，第二个线程必须调用“sem_post”的次数相等，以防止信号量溢出。此外，我担心信号量实现可能会在内部使用互斥量。

第二个不错的选择可能是使用signal，如果我能发现一种仅在第二个线程“清空队列并正在调用 sigwait”或“already waiting on sigwait”，简而言之，信号必须被提升最少次数，尽管如果信号被提升的次数比需要的多几次也没有坏处。是的，我确实使用了 Google 搜索，但我在 Internet 上找到的解决方案都不令人满意。以下是一些注意事项:

一个。服务器在进行系统调用时必须浪费最少的 CPU 周期，并且系统调用的使用次数必须最少。

B.开销必须非常低，算法必须高效。

C.没有任何锁定。

我希望所有选项都摆在桌面上。

这是我共享有关我的服务器的信息的站点的链接，以便更好地了解功能和用途: www.wanhive.com

最佳答案

如果您需要尽快唤醒线程#2，忙等待是个好办法。事实上，这是通知一个处理器有关另一个处理器所做更改的最快方式。您需要在两端生成内存栅栏(一侧写栅栏，另一侧读栅栏)。但是只有当你的两个线程都在专用处理器上执行时，这个说法才成立。在这种情况下，不需要上下文切换，只需缓存一致性流量。

可以做一些改进。

1. 如果线程 #2 通常受 CPU 限制并忙于等待 - 它可能会受到调度程序的惩罚(至少在 Windows 和 Linux 上)。操作系统调度程序动态调整线程优先级以提高整体系统性能。它降低了消耗大量 CPU 时间的 CPU 绑定(bind)线程的优先级，以防止线程饥饿。您需要手动提高线程 #2 的优先级以防止出现这种情况。
2. 如果您拥有多核或多处理器机器，您最终会遇到处理器订阅不足的情况，并且您的应用程序将无法利用硬件并发性。您可以通过使用多个处理器线程(线程 #2)来缓解这种情况。

处理步骤的并行化。 有两种选择。

1. 您的消息是完全有序的，需要按照它们到达时的相同顺序进行处理。
2. 消息可以重新排序。处理可以按任何顺序进行。

在第一种情况下，您需要 N 个循环缓冲区和 N 个处理线程以及 N 个输出缓冲区和一个消费者线程。线程 #1 在该循环缓冲区中按循环顺序对消息进行排队。

// Thread #1 pseudocode
auto message = recv()
auto buffer_index = atomic_increment(&message_counter);
buffer_index = buffer_index % N;  // N is the number of threads
// buffers is an array of cyclic buffers - Buffer* buffers[N];
Buffer* current_buffer = buffers[buffer_index];
current_buffer->euqueue(message);

每个线程都使用来自其中一个缓冲区的消息并将结果排队到他的专用输出缓冲区。

// Thread #i pseudocode
auto message = my_buffer->dequeue();
auto result = process(message);
my_output_buffer->enqueue(result);

现在您需要按到达顺序处理所有这些消息。您可以通过以循环方式从输出循环缓冲区中出列已处理的消息，使用专用的消费者线程来执行此操作。

// Consumer thread pseudocode
// out_message_counter is equal to message_counter at start
auto out_buffer_index = atomic_increment(&out_message_counter);
out_buffer_index = out_buffer_index % N;
// out_buffers is array of output buffers that is used by processing
// threads
auto out_buffer = out_buffers[out_buffer_index];
auto result = out_buffer->dequeue();
send(result);  // or whatever you need to do with result

在第二种情况下，当您不需要保留消息顺序时 - 您不需要消费者线程和输出循环缓冲区。您只需在处理线程中对结果做任何您需要做的事情。

N 必须等于 CPU 数量 - 第一种情况下为 3(“- 3”是一个 I/O 线程 + 一个消费者线程 + 一个 DHT 线程)和 CPU 数量 - 2 在第二种情况下(“- 2”是一个 I/O 线程 + 一个 DHT 线程)。这是因为如果处理器超额订阅，忙等待将无法生效。

关于c++ - 高吞吐量非阻塞服务器设计 : Alternatives to busy wait，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/20322422/