跳至主内容

RabbitMQ Stream 教程 - “Hello World!”

简介​

信息

先决条件​

本教程假定 RabbitMQ 已安装，正在localhost上运行，并且stream 插件已启用。标准的 stream 端口是 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:4-management

等待服务器启动，然后启用 stream 和 stream management 插件

docker exec rabbitmq rabbitmq-plugins enable rabbitmq_stream rabbitmq_stream_management

哪里寻求帮助​

如果您在学习本教程时遇到困难，可以通过邮件列表或Discord 社区服务器与我们联系。

RabbitMQ Streams 在 RabbitMQ 3.9 中引入。更多信息请参见此处。

"Hello World"​

（使用 .NET/C# Stream 客户端）​

在本教程的这一部分，我们将用 C# 编写两个程序：一个发送单个消息的生产者，一个接收消息并将其打印出来的消费者。我们将忽略 .NET 客户端 API 中的一些细节，专注于这个非常简单的入门示例。这是 RabbitMQ Streams 的“Hello World”。

.NET Stream 客户端库​

RabbitMQ 支持多种协议。本教程使用 RabbitMQ stream 协议，这是一个专用于RabbitMQ streams的协议。RabbitMQ 在许多不同语言中都有客户端，请参阅每种语言的 stream 客户端库。我们将使用 RabbitMQ 提供的.NET stream 客户端。

该客户端支持.NET。本教程将使用 RabbitMQ .NET stream 客户端 1.8.0 和 .NET，请确保您已安装并将其添加到您的 PATH 中。

RabbitMQ .NET stream 客户端 1.8 及更高版本通过nuget分发。

本教程假定您在 Windows 上使用 PowerShell。在 MacOS 和 Linux 上，几乎任何 shell 都可以工作。

设置​

首先，让我们验证一下您的 .NET 工具链是否在 PATH 中

dotnet --help

运行该命令应该会产生一个帮助消息。

可以在RabbitMQ 教程仓库中找到本教程的可执行版本。

现在，让我们生成两个项目，一个用于发布者，一个用于订阅者

dotnet new console --name Send
mv Send/Program.cs Send/Send.cs
dotnet new console --name Receive
mv Receive/Program.cs Receive/Receive.cs

这将创建两个名为Send和Receive的新目录。

然后，我们添加客户端依赖项。

cd Send
dotnet add package RabbitMQ.Stream.Client 
cd ../Receive
dotnet add package RabbitMQ.Stream.Client 
cd ..

现在 .NET 项目已设置好，我们可以编写一些代码了。

发送​

我们将消息生产者（发送者）命名为Send.cs，将消息消费者（接收者）命名为Receive.cs。生产者将连接到 RabbitMQ，发送一条消息，然后退出。

在Send.cs中，我们需要使用一些命名空间

using System.Text;
using RabbitMQ.Stream.Client;
using RabbitMQ.Stream.Client.Reliable;

然后，我们可以创建一个到服务器的连接

var streamSystem = await StreamSystem.Create(new StreamSystemConfig());

Stream .NET 客户端的入口点是StreamSystem。它用于配置 RabbitMQ stream 发布者、stream 订阅者和 stream 本身。

它封装了套接字连接，并为我们处理协议版本协商和身份验证等事宜。

本教程假定 stream 发布者和订阅者连接到本地运行的 RabbitMQ 节点，即连接到*localhost*。要连接到不同机器上的节点，只需在StreamSystemConfig中指定目标主机名或 IP 地址。

接下来，让我们创建一个生产者。

生产者还将声明一个它将发布消息的 stream，然后发布一条消息

await streamSystem.CreateStream(new StreamSpec("hello-stream")
{
    MaxLengthBytes = 5_000_000_000
});

var producer = await Producer.Create(new ProducerConfig(streamSystem, "hello-stream"));

await producer.Send(new Message(Encoding.UTF8.GetBytes($"Hello, World")));

stream 声明操作是幂等的：只有当 stream 不存在时才会创建它。

stream 是一个追加式日志抽象，允许在消息过期之前反复消费。定义保留策略是一个好习惯。在上面的示例中，stream 的大小限制为 5 GiB。

消息内容是字节数组。应用程序可以使用任何合适的格式（如 JSON、MessagePack 等）来编码需要传输的数据。

当上面的代码运行完毕后，生产者连接和 stream-system 连接将被关闭。我们的生产者就完成了。

每次运行生产者时，它都会向服务器发送一条消息，该消息将被追加到 stream 中。

完整的Send.cs 文件可以在 GitHub 上找到。

发送不起作用！​

如果这是您第一次使用 RabbitMQ，但没有看到“Sent”消息，您可能会感到困惑，不知道哪里出了问题。也许代理启动时磁盘空间不足（默认需要至少 50 MB 可用空间），因此拒绝接收消息。检查代理的日志文件，看看是否有资源警报已记录，并在必要时降低可用磁盘空间阈值。配置指南将展示如何设置 disk_free_limit。

另一个原因可能是程序在消息发送到代理之前就退出了。在某些客户端库中，发送是异步的：函数立即返回，但消息在发送到网络之前被放入 IO 层进行排队。发送程序要求用户按键完成进程：这样消息就有充足的时间到达代理。Stream 协议提供了一个确认机制来确保代理接收到出站消息，但为了简单起见，本教程不使用此机制。

接收​

本教程的另一部分，即消费者，将连接到 RabbitMQ 节点并等待消息被推送到它。与本教程中的生产者不同，生产者发送一条消息然后停止，消费者将持续运行，消费 RabbitMQ 推送给它的消息，并打印收到的负载。

与Send.cs类似，Receive.cs也需要使用一些命名空间

using System.Text;
using RabbitMQ.Stream.Client;
using RabbitMQ.Stream.Client.Reliable;

在初始设置方面，消费者部分与生产者部分非常相似；我们使用默认的连接设置，并声明消费者将从中消费的 stream。

请注意，stream 名称必须与生产者使用的名称匹配。

var streamSystem = await StreamSystem.Create(new StreamSystemConfig());

await streamSystem.CreateStream(new StreamSpec("hello-stream")
{
    MaxLengthBytes = 5_000_000_000
});

请注意，消费者部分也声明了 stream。这是为了允许任何一方先启动，无论是生产者还是消费者。

Consumer类用于实例化 stream 消费者，ConsumerConfig记录用于配置它。我们提供了一个MessageHandler回调来处理传递的消息。

OffsetSpec属性定义了消费者的起始点。在这种情况下，消费者从 stream 中可用的第一条消息开始。

var consumer = await Consumer.Create(new ConsumerConfig(streamSystem, "hello-stream")
{
    OffsetSpec = new OffsetTypeFirst(),
    MessageHandler = async (stream, _, _, message) =>
    {
        Console.WriteLine($"Stream: {stream} - " +
                          $"Received message: {Encoding.UTF8.GetString(message.Data.Contents)}");
        await Task.CompletedTask;
    }
});

完整的Receive.cs 文件可以在 GitHub 上找到。

整合​

要运行这两个示例，请打开两个终端（shell）标签页。

本教程的两个部分可以按任何顺序运行，因为它们都声明了 stream。让我们先运行消费者，这样当第一个发布者启动时，消费者就会打印出它

cd Receive
dotnet run

然后运行发布者

cd Send
dotnet run

消费者将打印它通过 RabbitMQ 从发布者那里收到的消息。消费者将继续运行，等待新的交付。尝试多次重新运行发布者来观察这一点。

Streams 与队列不同，它们是消息的追加式日志，可以被反复消费。当多个消费者从一个 stream 消费时，它们将从第一条可用消息开始。

© . This site is unofficial and not affiliated with VMware.