RabbitMQ Streams 概述

RabbitMQ 3.9 引入了一种新的数据结构：streams。Streams 能够解锁一系列使用“传统”队列实现起来可能很繁琐的用例。在这篇文章中，让我们一起探索 Streams 如何扩展 RabbitMQ 的功能。

什么是 RabbitMQ Streams

RabbitMQ 流（Stream）是一种具有非破坏性消费语义的仅追加日志（append-only log）。这意味着——与 RabbitMQ 中的传统队列不同——从流中消费消息并不会删除消息。

RabbitMQ 中的流是持久化且可复制的。这意味着数据安全和高可用性（以防节点丢失），同时也支持扩展（可从不同节点读取同一个流）。

与功能非常通用的传统队列相比，流的设计目标可能显得有些特定，但它们在某些场景下非常有用，并以非常好的方式扩展了 RabbitMQ 的功能。

流适用于什么场景

RabbitMQ Streams 在以下场景中表现尤为出色：

• 大规模扇出（Large fan-outs）：即多个应用程序需要读取相同的消息（使用传统队列，这需要为每个应用程序声明一个队列，并将消息的副本发送给它们中的每一个）
• 海量积压（Large backlogs）：流将消息存储在磁盘而非内存中，因此唯一的限制是磁盘容量
• 重放与时空穿梭（Replay & time-traveling）：消费者可以使用绝对偏移量或时间戳在流中的任何位置挂载，并且可以多次读取相同的数据
• 高吞吐量（High throughput）：与传统队列相比，流的速度极快，高出几个数量级

由于流作为 RabbitMQ 3.9 的一个核心插件提供，您可以将其与所有现有的 RabbitMQ 功能结合使用。

RabbitMQ Streams 简述

让我们进一步了解流的具体细节

• 得益于发布者确认（publisher confirms）和发布者端的重复数据删除，流提供了至少一次（at-least-once）的交付保证。
• 流支持服务器端的偏移量跟踪，让消费者可以在中断处恢复消费。
• 由于流具有非破坏性语义，数据量可能会增长很大。RabbitMQ Streams 可以根据大小或时间保留策略自动截断流。
• 可以通过专用的、极速的二进制协议访问流，也可以通过 AMQP 0.9.1 和 1.0（速度稍慢）进行访问。
• 流协议可以通过stream 插件访问，该插件包含在 RabbitMQ 3.9 的核心发行版中。
• RabbitMQ Streams 支持客户端-服务器 TLS 加密。
• 提供了一个现代化的、高度优化的 Java 客户端。它使用流协议以获得更好的性能，并已提供完整文档。
• 同时还提供了一个 Go 客户端。
• 还有一个基于 Java 客户端的性能测试工具。而且，它以 Docker 镜像的形式提供。

如果您想了解更多信息，可以查看下方 2021 年 RabbitMQ 峰会的流概述演示。如果您赶时间，可以跳过它直接进入下一节的 Docker 快速入门。

闲话少说，让我们开始运行吧。

Docker 快速入门

使用 Docker 体验流非常简单。首先确认您的本地没有我们要使用的 Docker 镜像：

docker rmi rabbitmq:3.9 pivotalrabbitmq/stream-perf-test

如果电脑上没有这些镜像，您会收到一条错误消息，但这没关系。

现在创建一个网络，以便我们的服务器和性能测试工具容器进行通信：

docker network create rabbitmq-streams

现在是时候启动 broker 了：

docker run -it --rm --network rabbitmq-streams --name rabbitmq rabbitmq:3.9

Broker 应该会在几秒钟内启动。当它准备就绪后，启用 stream 插件：

docker exec rabbitmq rabbitmq-plugins enable rabbitmq_stream

现在启动性能测试工具。它将创建一个流，并尽可能快地发布和消费消息：

docker run -it --rm --network rabbitmq-streams pivotalrabbitmq/stream-perf-test \
    --uris rabbitmq-stream://rabbitmq:5552

您可以让性能测试工具运行一段时间，然后使用 Ctrl+C 停止它：

19, published 1180489 msg/s, confirmed 1180145 msg/s, consumed 1180648 msg/s, \
    latency min/median/75th/95th/99th 1537/7819/9631/12136/14425 µs, chunk size 2639
20, published 1181929 msg/s, confirmed 1181597 msg/s, consumed 1182074 msg/s, \
    latency min/median/75th/95th/99th 1537/7838/9562/11967/14355 µs, chunk size 2657
^C
Summary: published 1205835 msg/s, confirmed 1205435 msg/s, consumed 1205477 msg/s, latency 95th 12158 µs, chunk size 2654

以上是在常规 Linux 工作站上的数据，具体数值取决于您的配置。请注意，在 macOS 和 Windows 上，数值可能会明显较低，因为 Docker 在这些操作系统上运行在虚拟化环境中。

然后，您可以使用 Ctrl+C 停止 broker 容器并删除该网络：

docker network rm rabbitmq-streams

如果您想进一步深入开发应用程序，Java 客户端文档是一个很好的起点。

以上就是 RabbitMQ Streams 的概述，这是一种具有出色功能和工具的全新仅追加日志数据结构。敬请关注后续文章，了解更多关于流的内容！