RabbitMQ Streams 的交付优化

RabbitMQ Streams 专为高吞吐量场景而设计，但当您的入口速率较低时会发生什么？较低的消息速率会严重影响交付性能，使消息消耗速率降低一个数量级。RabbitMQ 4.2 引入了一项优化，可显著提高低吞吐量流的交付速率，使所有支持的协议都受益。

利用 Streams 进行发布、存储与消费

当发布者应用程序使用流协议（stream protocol）向 RabbitMQ 流发送消息时，客户端库会将这些消息批量处理到一个发布帧（publishing frame）中。RabbitMQ 接收这些消息并进行聚合，通常会将来自多个发布者的消息合并为单个存储单元，即“数据块”（chunk）。随后，这个消息数据块会被持久化存储在文件系统中。

当消费者使用流协议订阅流时，RabbitMQ 会通过网络以顺序数据块的形式分发消息。每个数据块中包含的消息数量（即块大小）是影响吞吐量的关键因素；每个块包含数百条消息的大数据块，可以为消费者带来极高的消息投递速率。

接入速率（ingress rate）是决定块大小的主要因素：接入速率越高，块大小就越大。Streams 自设计之初就是为了实现高吞吐量。

小数据块带来的问题

当接入速率较低时会发生什么？此时每个数据块中仅包含少量消息，最坏的情况下甚至只有一条消息。这抵消了批处理带来的优势：每个帧只能向消费者投递极少量的消息。虽然数据块分发过程经过了优化，但它仍然需要多次系统调用（读取文件系统并写入套接字）。

虽然很难给出确切的数字，但假设一个流消费者在块大小为 300 时每秒可以从流中读取数百万条消息，而在块大小为 15 时，速率可能会下降到每秒约 20 万条消息。这里重要的是比例，而非绝对数值。

幸运的是，有一些方法可以改善小数据块带来的问题。

针对小数据块的优化：预读取（Read Ahead）

流往往具有一致的结构：如果一个流在某一点包含小数据块，那么它很可能大部分都由小数据块组成。那么，当我们读取到一个小数据块时，为什么不尝试预先读取后续的几个数据块呢？虽然分发多个数据块仍然需要多个帧，但我们只需执行一次文件系统读取操作，从而节省了一些昂贵的系统调用。

预读取的限制为 4,096 字节，因此并非所有小块流都能从这种优化中受益。尽管如此，这个简单的想法显著提高了特定流的投递速率。更好的是，RabbitMQ 不仅在流协议中使用了此技术，还将其应用到了所有协议（包括 AMQP）中。

性能结果

我们使用了一个 3 节点集群和一台虚拟机，利用 PerfTest 和 Stream PerfTest 进行了测试。所有虚拟机均为 m7i.4xlarge AWS 实例。我们创建了流并填充了 300 万条消息，同时改变了每个流的消息大小和块大小。随后，我们使用 PerfTest（AMQP 0.9.1）和 Stream PerfTest（流协议）消费了所有消息。我们在 RabbitMQ 4.1.4 和 4.2.0-beta.3 版本上进行了对比测试。

我们使用专门为此编写的工具填充了流，以获得预期的块大小。以下是用于消费消息的性能测试工具命令：

# PerfTest
java -jar perf-test.jar --producers 0 --consumers 1 --predeclared --qos 200 \
    --stream-consumer-offset first --cmessages 3000000 --queue stream
# Stream PerfTest
java -jar stream-perf-test.jar --producers 0 --consumers 1 --offset first \
    --initial-credits 50 --no-latency --cmessages 3000000 --streams stream

12 字节消息，AMQP 0.9.1

即使 12 字节的消息非常小，我们也能看到预读取方法效果良好：对于块大小为 1 条消息的流，速度提升了 10 倍。对于每个块包含 384 条消息的流，速率也依然是原来的两倍。

12 字节消息，流协议

在使用流协议且块大小为 1 条消息时，我们实现了近 10 倍的增长（从 16,000 提升至 134,000 条消息/秒）。在达到每个块 128 条消息之前，预读取的性能表现更优。

其他消息大小的测试结果与此一致，我们将在最后讨论它们。

48 字节消息，AMQP 0.9.1

48 字节消息，流协议

256 字节消息，AMQP 0.9.1

256 字节消息，流协议

512 字节消息，AMQP 0.9.1

512 字节消息，流协议

1024 字节消息，AMQP 0.9.1

1024 字节消息，流协议

结果分析

流协议的消费速率如预期般得到了提升：它在小块流中表现更高，但在数据块达到预读取限制（4,096 字节）时保持不变。这一点在 1,024 字节的消息中尤为明显，块大小为 1 和 2（小于预读取限制）时速率提升，但块大小为 4 和 6（达到或超过了限制，因此预读取不再起作用）时速率大致相同。

AMQP 0.9.1 的趋势类似，但我们观察到即使在数据块超过预读取限制后，结果仍有改进。对于 AMQP 0.9.1（以及其他非流协议，如 AMQP 1.0），RabbitMQ 使用了一种类似迭代器的方法来分发消息。这种方法此前就已经使用了某种形式的预读取，但现在它变得更加积极，并且能更好地适应数据块的大小。这解释了 AMQP 0.9.1 获得的良好结果，我们也应能在其他非流协议中获得同样的效果。

最后细节

预读取优化默认开启，但可以通过将 stream.read_ahead 配置项设置为 false 来全局禁用。

大数据块仍以优化方式分发给消费者：块头部被读取到内存中，而块数据（消息）通过零拷贝（zero-copy）方式发送到套接字。

结论

RabbitMQ 4.2 中的预读取优化证明，有时“瞻前顾后”确实能带来回报——在所有协议中，为小块流带来了高达 10 倍的性能提升。谁能想到在从磁盘读取时稍微贪心一点，就能让消费者如此开心呢？