一些队列理论：吞吐量、延迟和带宽

你在 RabbitMQ 中有一个队列。有一些客户端正在从该队列消费。如果你根本不设置 QoS 设置（basic.qos），那么 RabbitMQ 会尽可能快地将队列中的所有消息推送到客户端，只要网络和客户端允许。消费者将在内存中膨胀，因为它们会在自己的 RAM 中缓冲所有消息。如果你询问 RabbitMQ，队列可能看起来是空的，但可能有数百万条消息未被确认，因为它们位于客户端中，随时准备被客户端应用程序处理。如果你添加一个新的消费者，队列中就没有消息可以发送给新消费者了。消息只是被现有客户端缓冲，并且可能需要很长时间才能被处理，即使有其他消费者变得可用，可以更快地处理这些消息。这很不理想。

因此，默认的 QoS prefetch 设置为客户端提供了一个无限缓冲区，这可能导致行为和性能不佳。但是，QoS prefetch 缓冲区大小应该设置为多少？目标是让消费者保持饱和工作，但要最小化客户端的缓冲区大小，以便更多消息保留在 RabbitMQ 的队列中，从而可供新消费者使用，或者在消费者空闲时发送给它们。

假设 Rabbit 从队列获取一条消息、放入网络并最终抵达消费者需要 50ms。客户端处理该消息需要 4ms。消费者处理完消息后，会向 Rabbit 发送一个 ack（确认），该确认信息在网络传输和 Rabbit 处理上又需要 50ms。因此，总往返时间（RTT）为 104ms。如果我们设置 QoS prefetch（预取）为 1，那么在本次往返完成之前，Rabbit 不会发送下一条消息。这样，客户端在每 104ms 内仅工作 4ms，利用率仅为 3.8%。我们显然希望它的利用率达到 100%。

如果我们用 总往返时间 / 客户端处理每条消息的时间，得到 104 / 4 = 26。如果将 QoS prefetch 设置为 26，就能解决这个问题：假设客户端缓存了 26 条消息，随时准备处理。（这是一个合理的假设：一旦你设置了 basic.qos 并从队列 consume，Rabbit 会根据 QoS 限制，尽可能多地向客户端发送队列中的消息。如果假设消息不大且带宽充足，Rabbit 发送消息的速度很可能快于客户端处理的速度。因此，基于“客户端缓冲区已满”的假设来进行计算是合理且简化的。）如果处理每条消息需要 4ms，那么处理整个缓冲区共需 26 * 4 = 104ms。前 4ms 是客户端处理第一条消息的时间。随后客户端发送 ack 并继续处理下一条消息。该 ack 需要 50ms 到达代理（broker）。代理随后向客户端发送一条新消息，这又需要 50ms。因此，当 104ms 过去，客户端处理完缓冲区时，来自代理的下一条消息刚好送达，随时可供处理。这样客户端就一直处于忙碌状态：增加 QoS prefetch 不会让处理变得更快；相反，我们应尽量减小缓冲区大小以降低客户端内的消息延迟：让消息在客户端缓存的时间不超过维持客户端满负荷工作所需的时间。事实上，客户端能够完全清空缓冲区，直到下一条消息到达，因此缓冲区实际上一直保持为空。

只要处理时间和网络行为保持不变，这个方案非常完美。但考虑一下如果网络速度突然减半会发生什么：你的 prefetch 缓冲区就不够大了，客户端将会空闲，等待新消息到达，因为客户端的处理速度快于 Rabbit 的供应速度。

为了解决这个问题，我们可能会决定将 QoS prefetch 大小加倍（或接近加倍）。如果从 26 增加到 51，且客户端处理速度保持 4ms 每条，那么缓冲区内现在积压了 51 * 4 = 204ms 的消息，其中 4ms 用于处理当前消息，剩下的 200ms 用于将 ack 发回 Rabbit 并接收下一条消息。这样，我们就能应对网络速度减半的情况。

但是，如果网络表现正常，将 QoS prefetch 加倍意味着每条消息会在客户端缓冲区内停留一段时间，而不是一到达就立即处理。同样，从 51 条消息的满缓冲区开始，我们知道在客户端处理完第一条消息 100ms 后，新消息才会开始到达。但在那 100ms 内，客户端已经处理了 100 / 4 = 25 条消息。这意味着当新消息到达时，它会被添加到缓冲区末尾，而客户端从缓冲区头部移除消息。缓冲区因此将始终保持 50 - 25 = 25 条消息的长度，每条消息将在缓冲区中停留 25 * 4 = 100ms，这使得从 Rabbit 发送到客户端到客户端开始处理的时间延迟从 50ms 增加到了 150ms。

因此我们可以看出，为了让客户端在网络状况变差时仍能保持忙碌而增加 prefetch 缓冲区，会在网络状况良好时显著增加延迟。

同样，如果不是网络表现恶化，而是客户端处理每条消息的时间从 4ms 变成了 40ms 会怎样？如果 Rabbit 里的队列之前处于平衡状态（即流入和流出速率一致），现在由于流出速率降至原来的十分之一，积压的消息将会迅速增加。你可能会决定增加更多消费者来处理积压，但现有的客户端已经缓存了消息。假设原始缓冲区大小为 26 条消息，客户端将花费 40ms 处理第一条消息，然后发 ack 给 Rabbit 并转向下一条。ack 仍需 50ms 到达 Rabbit，Rabbit 再需 50ms 发出新消息，但在那 100ms 内，客户端只处理了 100 / 40 = 2.5 条后续消息，而不是剩下的 25 条。此时缓冲区积压了 25 - 3 = 22 条消息。从 Rabbit 发来的新消息不再会被立即处理，而是排在第 23 位，位于 22 条等待处理的消息之后，客户端在接下来的 22 * 40 = 880ms 内都不会处理它。鉴于 Rabbit 到客户端的网络延迟仅为 50ms，这额外的 880ms 延迟占了总延迟的 95%（880 / (880 + 50) = 0.946）。

更糟糕的是，如果我们为了应对网络性能下降将缓冲区加倍到 51 条消息呢？处理完第一条消息后，客户端还会缓存 50 条消息。100ms 后（假设网络正常），一条新消息从 Rabbit 到达，此时客户端刚好处理到缓冲区中 50 条消息里的第 3 条（缓冲区此时还剩 47 条），因此新消息在缓冲区中排在第 48 位，在接下来的 47 * 40 = 1880ms 内都不会被处理。同样，鉴于消息到达客户端的网络延迟只有 50ms，这额外的 1880ms 意味着客户端侧的缓冲导致了 97% 以上的延迟（1880 / (1880 + 50) = 0.974）。这很可能是不可接受的：数据只有在被及时处理时才有用，而不是在客户端收到它 2 秒后才处理！如果其他消费者处于空闲状态，它们也无能为力：一旦 Rabbit 将消息发送给某个客户端，该消息就由该客户端负责，直到它发送 ack 或 reject。消息一旦发送给某个客户端，其他客户端就无法“窃取”它。你真正想要的是：保持客户端忙碌，但让它们尽量少缓存消息，从而避免消息因客户端缓冲区而延迟，并让 Rabbit 队列中的消息能快速分配给新的消费者。

所以，缓冲区太小会导致网络变慢时客户端空闲，而缓冲区太大则会导致网络正常时出现大量额外延迟，并在客户端处理变慢时导致巨大的额外延迟。很明显，你需要的是一个可变的缓冲区大小。这些问题在网络设备中很常见，也是许多研究的主题。主动队列管理（Active Queue Management）算法试图通过丢弃或拒绝消息，来避免消息在缓冲区中长时间停留。当缓冲区保持为空时（每条消息仅受网络延迟影响，完全不滞留在缓冲区中），可以获得最低的延迟，而缓冲区仅用于吸收峰值。Jim Gettys 一直在从网络路由器的角度研究这个问题：局域网和广域网性能差异会遇到完全相同的问题。事实上，每当你在生产者（本例中为 Rabbit）和消费者（客户端逻辑）之间存在缓冲区，且两者的性能都可以动态变化时，你就会遇到这些问题。最近，一种称为 Controlled Delay（CoDel） 的新算法发表了，它在解决这些问题方面表现良好。

作者声称他们的 CoDel 算法是“免调节”的。这其实稍微有点夸大：它有两个旋钮需要适当地设置。但它们不需要每次性能变化时都去调整，这是一个巨大的优势。我已经为我们的 AMQP Java 客户端实现了这个算法，作为 QueueingConsumer 的一个变体。虽然原始算法针对的是 TCP 层，在 TCP 层丢弃数据包是合理的（TCP 本身会处理丢包重传），但在 AMQP 中这样做不太礼貌！因此，我的实现使用了 Rabbit 的 basic.nack 扩展，明确地将消息返回队列，以便其他消费者可以处理它们。

它的使用方法与普通的 QueueingConsumer 基本相同，只是为了获得最佳性能，你应该在构造函数中提供三个额外的参数。

1. 第一个是 requeue，它决定了当消息被 nack 时，应该重新入队还是丢弃。如果为 false，消息将被丢弃，如果已配置，则可能触发死信交换机机制。
2. 第二个是 targetDelay，它是消息在客户端侧 QoS prefetch 缓冲区中等待的可接受毫秒数。
3. 第三个是 interval，它是预期单条消息处理的最坏情况时间（毫秒）。这个值不需要非常精确，但保持在一个数量级内会有很大帮助。

你仍然应该适当地设置 QoS prefetch 大小。如果不设置，很可能客户端会被发送大量的消息，如果消息在缓冲区里呆得太久，算法就不得不把它们返还给 Rabbit。这样很容易导致消息返还给 Rabbit 时产生大量额外的网络流量。CoDel 算法旨在仅在性能偏离常态时才开始丢弃（或拒绝）消息，因此一个具体的例子会有帮助。

再次假设每个方向的网络传输时间为 50ms，我们预期客户端处理每条消息平均需要 4ms，但可能会飙升至 20ms。因此我们将 CoDel 的 interval 参数设为 20。有时网络速度减半，传输时间变为双向各 100ms。为了适应这种情况，我们将 basic.qos prefetch 设置为 204 / 4 = 51。是的，这意味着当网络正常运行时，缓冲区大部分时间会保持 25 条消息的长度（参见之前的计算），但我们认为这是可以接受的。每条消息预计会在缓冲区中停留 25 * 4 = 100ms，因此我们将 CoDel 的 targetDelay 设为 100。

当一切运行正常时，CoDel 不应介入，几乎不会有消息被 nack。但如果客户端开始比平时更慢地处理消息，CoDel 将发现消息在客户端缓存的时间太长，并将这些消息返还给队列。如果这些消息被重新入队，它们将可以发送给其他客户端。

这目前还是非常实验性的，我们完全可以找到理由说明为什么 CoDel 处理 AMQP 消息不像处理普通 IP 数据包那样直接。还需要记住的是，通过 nack 重新入队消息是一个相当昂贵的操作，因此最好设置 CoDel 参数以确保在正常运行中几乎没有消息被 nack。管理插件是查看有多少消息被 nack 的简单方法。一如既往，欢迎任何评论、反馈和改进建议！