RabbitMQ 流概述

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

什么是 RabbitMQ 流

RabbitMQ 流模拟一个具有非破坏性消费语义的追加式日志。这意味着与 RabbitMQ 中的传统队列相反，从流中消费不会删除消息。

RabbitMQ 中的流是持久化和复制的。这意味着数据安全性和可用性（在节点丢失的情况下），以及可扩展性（从不同节点读取相同的流）。

与用途广泛的队列相比，流可能看起来有些固执己见，但它们在某些用例中非常有用。它们以一种非常不错的方式扩展了 RabbitMQ 的功能。

流适合什么

RabbitMQ 流在以下用例中表现出色

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

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

RabbitMQ 流概览

让我们更详细地了解流

• 流通过发布者确认和发布者端的消重提供至少一次保证。
• 流支持服务器端偏移量跟踪，以使消费者能够从上次停止的地方重新开始。
• 由于流具有非破坏性语义，因此它们可以增长很多。RabbitMQ 流可以根据大小或年龄，根据保留策略自动截断流。
• 可以通过专门的、超快的二进制协议以及通过 AMQP 0.9.1 和 1.0（速度较慢）访问流。
• 可以通过流插件访问流协议，该插件包含在 RabbitMQ 3.9 的核心发行版中。
• RabbitMQ 流支持客户端-服务器 TLS。
• 有一个现代的、高度优化的Java 客户端可用。它使用流协议以获得更好的性能。它已完全记录。
• 还提供了一个Go 客户端。
• 还有一个基于 Java 客户端的性能工具。是的，它作为一个Docker 镜像提供。

如果您想了解更多信息，可以查看下方来自 RabbitMQ Summit 2021 的流概述演示文稿。如果您赶时间，可以跳过它，直接进入下一节中使用 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 流的概述，这是一种具有强大功能和工具的新追加式日志数据结构。请继续关注，了解后续文章中关于流的更多信息！