RabbitMQ 流教程 - “Hello World！”
简介
先决条件
本教程假设 RabbitMQ 已安装，并在 localhost 上运行，并且流插件已启用。标准流端口为 5552。如果您使用不同的主机、端口或凭据，则连接设置需要进行调整。
使用 Docker
如果您没有安装 RabbitMQ，可以在 Docker 容器中运行它
docker run -it --rm --name rabbitmq -p 5552:5552 -p 15672:15672 -p 5672:5672 \
-e RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS='-rabbitmq_stream advertised_host localhost' \
rabbitmq:3.13
等待服务器启动，然后启用流和流管理插件
docker exec rabbitmq rabbitmq-plugins enable rabbitmq_stream rabbitmq_stream_management
获取帮助
如果您在学习本教程时遇到问题，可以通过邮件列表或Discord 社区服务器联系我们。
RabbitMQ Streams 在 RabbitMQ 3.9 中引入。更多信息请访问此处。
"Hello World"
(使用 Python (rstream) 流客户端)
在本教程的这一部分，我们将用 Python 编写两个程序；一个生产者发送一条消息，一个消费者接收消息并打印出来。我们将略过 Python 客户端 API 中的一些细节，专注于这个非常简单的事情，以便开始。这是 RabbitMQ Streams 的“Hello World”。
Python (rstream) 流客户端库
RabbitMQ 支持多种协议。本教程使用 RabbitMQ 流协议，这是一个专用于RabbitMQ 流的协议。许多不同语言中都有大量 RabbitMQ 客户端，请参阅每种语言的流客户端库。我们将使用由 George Fortunatov 构建的、现在由 RabbitMQ 支持的Python (rstream) 流客户端。
该客户端支持Python >= 3.9。本教程将使用 rstream 客户端 0.19.1 版本。Python (rstream) 客户端 0.19.1 及更高版本通过pip分发。
本教程假设您在 Windows 上使用 PowerShell。在 macOS 和 Linux 上，几乎任何 Shell 都可以使用。
设置
首先，让我们验证您的 PATH 中是否包含 Python 工具链
python --help
运行该命令应该会生成一条帮助消息。
本教程的可执行版本可在RabbitMQ 教程存储库中找到。
现在让我们创建一个文件夹项目并安装依赖项
# using pip
mkdir python-rstream
cd python-rstream
pip install rstream
# using Pipenv
mkdir python-rstream
cd python-rstream
pipenv install rstream
pipenv shell
现在创建两个名为 send.py 和 receive.py 的新文件。现在我们已经设置了 Python 项目，我们可以编写一些代码了。
发送
我们将消息生产者（发送者）称为 send.py，消息消费者（接收者）称为 receive.py。生产者将连接到 RabbitMQ，发送一条消息，然后退出。
在send.py中，我们需要一些导入
import asyncio
from rstream import Producer
然后我们可以创建到服务器的连接
async with Producer(
host="localhost",
username="guest",
password="guest",
) as producer
生产者的入口点是 Producer 类。它用于配置 RabbitMQ 流发布者和流本身。
它抽象了套接字连接，并为我们处理协议版本协商和身份验证等。
本教程假设流发布者和消费者连接到本地运行的 RabbitMQ 节点，即 localhost。要连接到其他机器上的节点，只需在 Producers 参数中指定目标主机名或 IP 地址即可。
接下来，让我们创建一个生产者。
生产者还将声明它将向其发布消息的流，然后发布一条消息
STREAM_NAME = "hello-python-stream"
# 5GB
STREAM_RETENTION = 5000000000
await producer.create_stream(
STREAM_NAME, exists_ok=True, arguments={"MaxLengthBytes": STREAM_RETENTION})
await producer.send(stream=STREAM_NAME, message=b"Hello, World!")
流声明操作是幂等的：只有在流不存在时才会创建它。
流是一种追加式日志抽象，允许重复使用消息，直到它们过期。始终定义保留策略是一个好习惯。在上面的示例中，流的大小限制为 5 GiB。
消息内容是字节数组。应用程序可以使用任何合适的格式（如 JSON、MessagePack 等）对它们需要传输的数据进行编码。
当上面的代码执行完毕后，生产者连接将关闭。这就是我们的生产者。
每次运行生产者时，它都会向服务器发送一条消息，并且该消息将附加到流中。
完整的send.py 文件可以在 GitHub 上找到。
发送不起作用！
如果这是您第一次使用 RabbitMQ，并且您没有看到“已发送”消息，那么您可能会挠头想知道哪里出了问题。也许代理在启动时没有足够的可用磁盘空间（默认情况下需要至少 50 MB 的可用空间），因此拒绝接受消息。检查代理日志文件，查看是否记录了资源警报，如有必要，降低可用磁盘空间阈值。配置指南将向您展示如何设置
disk_free_limit。另一个原因可能是程序在消息到达代理之前就退出了。在某些客户端库中，发送是异步的：函数立即返回，但消息在通过网络发送之前会排队到 IO 层。发送程序要求用户按一个键以完成该过程：消息有足够的时间到达代理。流协议提供了一种确认机制来确保代理接收出站消息，但为了简单起见，本教程没有使用此机制。
接收
本教程的另一部分，即消费者，将连接到 RabbitMQ 节点并等待消息被推送到它。与本教程中发布一条消息并停止的生产者不同，消费者将持续运行，使用 RabbitMQ 推送到它的消息，并打印收到的有效负载。
与 send.py 类似，receive.py 需要一些导入
import asyncio
import signal
from rstream import (
AMQPMessage,
Consumer,
MessageContext,
ConsumerOffsetSpecification,
OffsetType
)
在初始设置方面，消费者部分与生产者部分非常相似；我们使用默认连接设置并声明消费者将从中消费的流。
consumer = Consumer(host="localhost", username="guest", password="guest")
await consumer.create_stream(
STREAM_NAME, exists_ok=True, arguments={"MaxLengthBytes": STREAM_RETENTION}
)
请注意，消费者部分也声明了流。这是为了允许生产者或消费者中的任何一方首先启动。
我们向 consumer.subscribe 函数提供了一个 on_message 回调。
offset_specification 定义了消费者的起点。在这种情况下，消费者从流中第一个可用的消息开始。
async def on_message(msg: AMQPMessage, message_context: MessageContext):
stream = message_context.consumer.get_stream(message_context.subscriber_name)
print("Got message: {} from stream {}".format(msg, stream))
await consumer.start()
await consumer.subscribe(
stream=STREAM_NAME,
callback=on_message,
offset_specification=ConsumerOffsetSpecification(OffsetType.FIRST, None),
)
await consumer.run()
完整的receive.py 文件可以在 GitHub 上找到。
综合示例
为了运行这两个示例，请打开两个终端（shell）选项卡。
本教程的两个部分可以按任何顺序运行，因为它们都声明了流。让我们先运行消费者，这样当第一个发布者启动时，消费者将打印它
python receive.py
然后运行生产者
python send.py
消费者将打印它通过 RabbitMQ 从发布者接收到的消息。消费者将继续运行，等待新的传递。尝试重新运行发布者几次以观察这一点。
流与队列的不同之处在于，它们是可重复使用的消息的追加式日志。当多个消费者从流中消费时，它们将从第一个可用的消息开始。