跳至主内容

RabbitMQ Stream 教程 - 偏移量跟踪

简介​

信息

先决条件​

本教程假定 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 中引入。更多信息请参见此处。

偏移量跟踪​

设置​

本教程的这部分将编写两个 C# 程序：一个生产者，它发送一系列带有标记消息的消息，以及一个消费者，它接收消息并在收到标记消息时停止。它展示了消费者如何浏览流，甚至可以从之前的执行中断处重新开始。

本教程使用了 Stream .NET 客户端。请确保按照第一个教程中的 设置步骤 进行操作。

本教程的可执行版本可以在 RabbitMQ 教程存储库 中找到。发送程序名为 OffsetTrackingSend.cs，接收程序名为 OffsetTrackingReceive.cs。

让我们为每个程序创建一个项目

dotnet new console --name OffsetTrackingSend
mv OffsetTrackingSend/Program.cs OffsetTrackingSend/OffsetTrackingSend.cs
dotnet new console --name OffsetTrackingReceive
mv OffsetTrackingReceive/Program.cs OffsetTrackingReceive/OffsetTrackingReceive.cs

然后添加客户端依赖。

cd OffsetTrackingSend
dotnet add package RabbitMQ.Stream.Client 
cd ../OffsetTrackingReceive
dotnet add package RabbitMQ.Stream.Client 
cd ..

本教程侧重于客户端库的使用，因此存储库中的最终代码应用于创建文件的脚手架（例如，导入）。

发送​

发送程序首先创建 StreamSystem 并声明流

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

var stream = "stream-offset-tracking-dotnet";
await streamSystem.CreateStream(new StreamSpec(stream));

然后，程序创建一个 Producer 实例并发布 100 条消息。最后一条消息的正文值设置为 marker；这是消费者停止消费的标记。

注意 CountdownEvent 的使用：它在每个消息确认回调中通过 Signal() 递减。这确保了在程序关闭之前，代理已收到所有消息。

var messageCount = 100;
var confirmedCde = new CountdownEvent(messageCount);
var producer = await Producer.Create(new ProducerConfig(streamSystem, stream) {
    ConfirmationHandler = async confirmation => {
        if (confirmation.Status == ConfirmationStatus.Confirmed) {
            confirmedCde.Signal();
        }
        await Task.CompletedTask.ConfigureAwait(false);
    }
});

Console.WriteLine("Publishing {0} messages...", messageCount);
for (int i = 0; i < messageCount; i++) {
    var body = i == messageCount - 1 ? "marker" : "hello";
    await producer.Send(new Message(Encoding.UTF8.GetBytes(body)));
}

confirmedCde.Wait();
Console.WriteLine("Messages confirmed.");
await producer.Close();
await streamSystem.Close();

现在，让我们创建接收程序。

接收​

接收程序创建一个 StreamSystem 实例，并确保流也被创建。代码的这部分与发送程序相同，因此为了简洁起见，在接下来的代码片段中被省略了。

接收程序启动一个附加到流开头的消费者（new OffsetTypeFirst()）。它使用变量在程序结束时输出第一个和最后一个接收到的消息的偏移量。

消费者在收到标记消息时停止：它将偏移量分配给一个变量，关闭消费者，并递减 CountdownEvent。与发送者一样，当消费者完成其工作时，CountdownEvent 会告知程序继续。

IOffsetType offsetSpecification = new OffsetTypeFirst();
ulong initialValue = UInt64.MaxValue;
ulong firstOffset = initialValue;
ulong lastOffset = initialValue;
var consumedCde = new CountdownEvent(1);
var consumer = await Consumer.Create(new ConsumerConfig(streamSystem, stream)
{
    OffsetSpec = offsetSpecification,
    MessageHandler = async (_, consumer, context, message) => {
        if (Interlocked.CompareExchange(ref firstOffset, context.Offset, initialValue) == initialValue) {
            Console.WriteLine("First message received.");
        }
        if ("marker".Equals(Encoding.UTF8.GetString(message.Data.Contents))) {
            Interlocked.Exchange(ref lastOffset, context.Offset);
            await consumer.Close();
            consumedCde.Signal();
        }
        await Task.CompletedTask;
    }
});
Console.WriteLine("Started consuming...");

consumedCde.Wait();
Console.WriteLine("Done consuming, first offset {0}, last offset {1}.", firstOffset, lastOffset);
await streamSystem.Close();

探索流​

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

在第一个选项卡中，进入 OffsetTrackingSend 目录，然后运行发送程序发布一系列消息

dotnet run

输出如下：

Publishing 100 messages...
Messages confirmed.

现在让我们运行接收程序。打开一个新的选项卡，进入 OffsetTrackingSend 目录。请记住，由于指定了 first 偏移量，它应该从流的开头开始。

dotnet run

这是输出：

Started consuming...
First message received.
Done consuming, first offset 0, last offset 99
什么是偏移量？

流可以看作是一个包含消息的数组。偏移量是数组中给定消息的索引。

流与队列不同：消费者可以读取和重读相同的消息，并且消息会保留在流中。

让我们通过使用 offset(long) 规范附加到给定偏移量来尝试此功能。将 offsetSpecification 变量从 OffsetTypeFirst() 设置为 OffsetTypeOffset(42)

IOffsetType offsetSpecification = new OffsetTypeOffset(42);

偏移量 42 是任意的，它可以是 0 到 99 之间的任何数字。再次运行接收程序。

dotnet run

输出如下：

Started consuming...
First message received.
Done consuming, first offset 42, last offset 99.

还有一种方法可以附加到流的末尾，以便在创建消费者时只查看新消息。这就是 next 偏移量规范。让我们试试。

IOffsetType offsetSpecification = new OffsetTypeNext();

运行接收程序。

dotnet run

这次消费者没有收到任何消息。

Started consuming...

它正在等待流中的新消息。通过再次运行发送程序来发布一些消息。回到第一个选项卡。

dotnet run

等待程序退出，然后切换回接收程序选项卡。消费者收到了新消息。

Started consuming...
First message received.
Done consuming, first offset 100, last offset 199.

接收程序因为发送程序将其放在流末尾的新标记消息而停止。

本节展示了如何“浏览”流：从开头、从任何偏移量，甚至对于新消息。下一节将介绍如何利用服务器端偏移量跟踪，以便从消费者前一次执行的中断处恢复。

服务器端偏移量跟踪​

RabbitMQ Streams 提供服务器端偏移量跟踪，用于存储流中给定消费者的进度。如果消费者因任何原因停止（崩溃、升级等），它将能够从先前停止的位置重新连接，以避免处理相同的消息。

RabbitMQ Streams 提供了偏移量跟踪的 API，但也可以使用其他解决方案来存储正在消耗的应用程序的进度。这可能取决于用例，但关系型数据库也是一个不错的解决方案。

让我们修改接收程序以存储已处理消息的偏移量。已更新的行用注释标出。

var consumerName = "offset-tracking-tutorial"; // name of the consumer
IOffsetType offsetSpecification;
try {
    // get last stored offset
    ulong storedOffset = await streamSystem.QueryOffset(consumerName, stream).ConfigureAwait(false);
    // start just after the last stored offset
    offsetSpecification = new OffsetTypeOffset(storedOffset + 1);
} catch (OffsetNotFoundException) {
    // start consuming at the beginning of the stream if no stored offset
    offsetSpecification = new OffsetTypeFirst();
}
ulong initialValue = UInt64.MaxValue;
ulong firstOffset = initialValue;
int messageCount = 0; // number of received messages
ulong lastOffset = initialValue;
var consumedCde = new CountdownEvent(1);
var consumer = await Consumer.Create(new ConsumerConfig(streamSystem, stream)
{
    OffsetSpec = offsetSpecification,
    Reference =  consumerName,  // the consumer must a have name
    MessageHandler = async (_, consumer, context, message) => {
        if (Interlocked.CompareExchange(ref firstOffset, context.Offset, initialValue) == initialValue) {
            Console.WriteLine("First message received.");
        }
        if (Interlocked.Increment(ref messageCount) % 10 == 0) {
            // store offset every 10 messages
            await consumer.StoreOffset(context.Offset).ConfigureAwait(false);
        }
        if ("marker".Equals(Encoding.UTF8.GetString(message.Data.Contents))) {
            Interlocked.Exchange(ref lastOffset, context.Offset);
            // store the offset on consumer closing
            await consumer.StoreOffset(context.Offset).ConfigureAwait(false);
            await consumer.Close();
            consumedCde.Signal();
        }
        await Task.CompletedTask;
    }
});
Console.WriteLine("Started consuming...");

consumedCde.Wait();
Console.WriteLine("Done consuming, first offset {0}, last offset {1}.", firstOffset, lastOffset);
await streamSystem.Close();

最重要的更改是

  • 程序在创建消费者之前查找最后存储的偏移量。如果没有存储的偏移量（这很可能是此消费者第一次启动），它会使用 first。如果存储了偏移量，它会使用 offset 规范紧随其后（存储的偏移量 + 1）启动，这假设具有存储偏移量的消息已在应用程序的先前实例中处理。
  • 消费者必须有一个名称。它是存储和检索最后一个存储的偏移量值的关键。
  • 偏移量每 10 条消息存储一次。对于偏移量存储频率而言，这是一个异常低的值，但对于本教程来说没问题。实际世界中的值通常是几百或几千。
  • 偏移量在关闭消费者之前存储，就在收到标记消息之后。

现在让我们运行更新后的接收程序。

dotnet run

这是输出：

Started consuming...
First message received.
Done consuming, first offset 0, last offset 99.

这没什么令人惊讶的：消费者从流的开头获取了消息，并在到达标记消息时停止。

让我们再次启动它。

dotnet run

这是输出：

Started consuming...
First message received.
Done consuming, first offset 100, last offset 199.

消费者正好从上次中断的地方恢复：第一次运行的最后一个偏移量是 99，第二次运行的第一个偏移量是 100。消费者在第一次运行时存储了偏移量跟踪信息，因此客户端库在第二次运行时使用它来恢复到正确的位置继续消费。

本教程关于 RabbitMQ Streams 中的消费语义的内容到此结束。它涵盖了消费者如何附加到流中的任何位置。消费应用程序很可能需要跟踪它们在流中达到的点。它们可以使用本教程中演示的内置服务器端偏移量跟踪功能。它们也可以自由使用任何其他数据存储解决方案来完成此任务。

有关偏移量跟踪的更多信息，请参阅 RabbitMQ 博客 和 Stream .NET 客户端文档。

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