RabbitMQ Streams 概述

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

RabbitMQ Streams 是什么

RabbitMQ Streams 将不可变日志与非破坏性消费语义相结合。这意味着，与 RabbitMQ 中的传统队列不同，从流中消费消息不会移除这些消息。

RabbitMQ 中的流是持久化和复制的。这带来了数据安全性和可用性（以防节点丢失），以及可扩展性（可以从不同节点读取同一流）。

与功能非常强大的队列相比，流可能显得有些固定，但它们在许多用例中非常有用。它们以一种非常好的方式扩展了 RabbitMQ 的功能。

流的优势是什么

RabbitMQ Streams 在以下用例中表现出色：

• 大型扇出：许多应用程序需要读取相同的消息（使用传统队列，这需要为每个应用程序声明一个队列，并将同一消息的副本发送给每个队列）
• 大型积压：流将消息存储在磁盘上，而不是内存中，因此唯一的限制是磁盘容量
• 重放与时间旅行：消费者可以附加到流中的任何位置，使用绝对偏移量或时间戳，并且可以读写相同的数据
• 高吞吐量：与传统队列相比，流的速度非常快，快了几个数量级

并且，由于流作为 RabbitMQ 3.9 的核心插件发布，您可以将其与所有现有的 RabbitMQ 功能一起使用。

RabbitMQ Streams 概览

让我们更具体地了解流

• 流通过发布者确认和发布者端的去重来提供至少一次的保证。
• 流支持服务器端偏移量跟踪，让消费者可以从上次离开的地方继续。
• 由于流具有非破坏性语义，它们可以增长很多。RabbitMQ Streams 可以根据基于大小或年龄的保留策略自动截断流。
• 流可以通过专用的、极快的二进制协议以及 AMQP 0.9.1 和 1.0（速度较慢）进行访问。
• 流协议可以通过流插件访问，该插件随 RabbitMQ 3.9 的核心发行版一起提供。
• RabbitMQ Streams 支持客户端-服务器 TLS。
• 提供了一个现代、高度优化的Java 客户端。它使用流协议以获得更好的性能。该文档完整。
• 还提供了一个Go 客户端。
• 还有一个基于 Java 客户端的性能工具。是的，它以Docker 镜像的形式提供。

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

废话不多说，让我们开始运行吧。

Docker 快速入门

使用 Docker 来体验流非常简单。让我们确保您本地还没有我们将要使用的 Docker 镜像

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

如果计算机上没有这些镜像，您将收到一个错误消息，但这没关系。

现在，让我们为我们的服务器和性能工具容器创建一个网络，以便它们可以通信

docker network create rabbitmq-streams

是时候启动代理了

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

代理应该会在几秒钟内启动。准备就绪后，启用流插件

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 停止代理容器并删除网络

docker network rm rabbitmq-streams

如果您想进一步构建应用程序，流 Java 客户端文档是一个不错的起点。

以上就是我们对 RabbitMQ Streams 的概述，它是一种具有强大功能和工具的新型不可变日志数据结构。请继续关注我们后续的帖子，了解更多关于流的信息！