RabbitMQ 流教程 - 偏移量跟踪
介绍
先决条件
本教程假设 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 流在 RabbitMQ 3.9 中引入。更多信息请访问 这里。
偏移量跟踪
设置
本教程的一部分是编写两个 Java 程序；一个生产者，它发送一系列消息，并在最后发送一个标记消息，以及一个消费者，它接收消息，并在收到标记消息时停止。它展示了消费者如何浏览流，甚至可以在先前执行停止的地方重新开始。
本教程使用 流 Java 客户端。请确保遵循 第一个教程的设置步骤。
本教程的可执行版本可以在 RabbitMQ 教程仓库 中找到。发送程序称为 OffsetTrackingSend.java，接收程序称为 OffsetTrackingReceive.java。本教程侧重于客户端库的使用，因此仓库中的最终代码应该用于创建文件的框架（例如，导入、主函数等）。
发送
发送程序首先实例化 Environment 并创建流
try (Environment environment = Environment.builder().build()) {
String stream = "stream-offset-tracking-java";
environment.streamCreator()
.stream(stream)
.maxLengthBytes(ByteCapacity.GB(1))
.create();
// publishing code to come
}
然后，程序创建 Producer 实例并发布 100 条消息。最后一条消息的主体值为 marker；这是消费者停止消费的标记。
请注意 CountDownLatch 的使用：它在每个消息确认回调中使用 countDown 进行递减。这确保了代理在关闭程序之前收到了所有消息。
Producer producer = environment.producerBuilder()
.stream(stream)
.build();
int messageCount = 100;
CountDownLatch confirmedLatch = new CountDownLatch(messageCount);
System.out.printf("Publishing %d messages%n", messageCount);
IntStream.range(0, messageCount).forEach(i -> {
String body = i == messageCount - 1 ? "marker" : "hello";
producer.send(producer.messageBuilder()
.addData(body.getBytes(UTF_8))
.build(),
ctx -> {
if (ctx.isConfirmed()) {
confirmedLatch.countDown();
}
}
);
});
boolean completed = confirmedLatch.await(60, TimeUnit.SECONDS);
System.out.printf("Messages confirmed: %b.%n", completed);
现在让我们创建接收程序。
接收
接收程序创建 Environment 实例，并确保流也已创建。这部分代码与发送程序相同，因此在接下来的代码片段中为了简洁起见，它被省略了。
接收程序启动一个消费者，该消费者在流的开头（OffsetSpecification.first()）附加。它使用变量在程序结束时输出第一个和最后一个接收到的消息的偏移量。
当消费者收到标记消息时，它会停止：它将偏移量分配给一个变量，关闭消费者，并递减闩锁计数。与发送方一样，CountDownLatch 告诉程序在消费者完成工作后继续执行。
OffsetSpecification offsetSpecification = OffsetSpecification.first();
AtomicLong firstOffset = new AtomicLong(-1);
AtomicLong lastOffset = new AtomicLong(0);
CountDownLatch consumedLatch = new CountDownLatch(1);
environment.consumerBuilder()
.stream(stream)
.offset(offsetSpecification)
.messageHandler((ctx, msg) -> {
if (firstOffset.compareAndSet(-1, ctx.offset())) {
System.out.println("First message received.");
}
String body = new String(msg.getBodyAsBinary(), StandardCharsets.UTF_8);
if ("marker".equals(body)) {
lastOffset.set(ctx.offset());
ctx.consumer().close();
consumedLatch.countDown();
}
})
.build();
System.out.println("Started consuming...");
consumedLatch.await(60, TimeUnit.MINUTES);
System.out.printf("Done consuming, first offset %d, last offset %d.%n",
firstOffset.get(), lastOffset.get());
探索流
为了运行这两个示例，请打开两个终端（shell）选项卡。
在第一个选项卡中，运行发送方以发布一系列消息
./mvnw -q compile exec:java '-Dexec.mainClass=OffsetTrackingSend'
输出如下
Publishing 100 messages...
Messages confirmed: true.
现在让我们运行接收方。打开一个新选项卡。请记住，由于 first 偏移量规范，它应该从流的开头开始。
./mvnw -q compile exec:java '-Dexec.mainClass=OffsetTrackingReceive'
以下是输出
Started consuming...
First message received.
Done consuming, first offset 0, last offset 99.
流可以看作是一个数组，其中元素是消息。偏移量是给定消息在数组中的索引。
流与队列不同：消费者可以读取和重新读取相同的消息，并且消息会保留在流中。
让我们尝试使用 offset(long) 规范在给定偏移量处附加来使用此功能。将 offsetSpecification 变量从 OffsetSpecification.first() 设置为 OffsetSpecification.offset(42)
OffsetSpecification offsetSpecification = OffsetSpecification.offset(42);
偏移量 42 是任意的，它可以是 0 到 99 之间的任何数字。再次运行接收方
./mvnw -q compile exec:java '-Dexec.mainClass=OffsetTrackingReceive'
输出如下
Started consuming...
First message received.
Done consuming, first offset 42, last offset 99.
还有一种方法可以在流的末尾附加，以便仅在创建消费者时查看新消息。这就是 next 偏移量规范。让我们尝试一下
OffsetSpecification offsetSpecification = OffsetSpecification.next();
运行接收方
./mvnw -q compile exec:java '-Dexec.mainClass=OffsetTrackingReceive'
这次消费者没有收到任何消息
Started consuming...
它正在等待流中的新消息。让我们通过再次运行发送方来发布一些消息。回到第一个选项卡
./mvnw -q compile exec:java '-Dexec.mainClass=OffsetTrackingSend'
等待程序退出，然后切换回接收方选项卡。消费者收到了新消息
Started consuming...
First message received.
Done consuming, first offset 100, last offset 199.
接收方由于发送方在流的末尾放置的新标记消息而停止。
本节展示了如何“浏览”流：从开头、从任何偏移量，甚至用于新消息。下一节介绍如何利用服务器端偏移量跟踪来恢复消费者在先前执行停止的地方。
服务器端偏移量跟踪
RabbitMQ 流提供服务器端偏移量跟踪，以存储给定消费者在流中的进度。如果消费者因任何原因停止（崩溃、升级等），它将能够重新附加到先前停止的位置，以避免处理相同的消息。
RabbitMQ 流提供了偏移量跟踪的 API，但可以使用其他解决方案来存储消费应用程序的进度。这可能取决于用例，但关系数据库也可以是一个不错的解决方案。
让我们修改接收方以存储已处理消息的偏移量。使用注释突出显示更新的行
// start consuming at the beginning of the stream
OffsetSpecification offsetSpecification = OffsetSpecification.first();
AtomicLong messageCount = new AtomicLong(0);
environment.consumerBuilder()
.stream(stream)
.offset(offsetSpecification)
.name("offset-tracking-tutorial") // the consumer must a have name
.manualTrackingStrategy().builder() // activate manual offset tracking
.messageHandler((ctx, msg) -> {
if (firstOffset.compareAndSet(-1, ctx.offset())) {
System.out.println("First message received.");
}
if (messageCount.incrementAndGet() % 10 == 0) {
ctx.storeOffset(); // store offset every 10 messages
}
String body = new String(msg.getBodyAsBinary(), StandardCharsets.UTF_8);
if (body.equals("marker")) {
lastOffset.set(ctx.offset());
ctx.storeOffset(); // store the offset on consumer closing
ctx.consumer().close();
consumedLatch.countDown();
}
})
.build();
最相关的更改是
- 消费者使用
OffsetSpecification.first()在流的开头附加。 - 消费者必须有一个名称。它是存储和检索最后存储的偏移量值的键。
- 手动跟踪策略已激活，这意味着需要显式调用以存储偏移量。
- 每 10 条消息存储一次偏移量。对于偏移量存储频率来说，这是一个异常低的价值，但这对于本教程来说是可以的。现实世界中的值通常在数百或数千范围内。
- 在关闭消费者之前，就在获取标记消息之后，存储偏移量。
让我们运行更新后的接收方
./mvnw -q compile exec:java '-Dexec.mainClass=OffsetTrackingReceive'
以下是输出
Started consuming...
First message received.
Done consuming, first offset 0, last offset 99.
这里没有什么令人惊讶的：消费者从流的开头获取了消息，并在到达标记消息时停止。
让我们再次启动它
./mvnw -q compile exec:java '-Dexec.mainClass=OffsetTrackingReceive'
以下是输出
Started consuming...
First message received.
Done consuming, first offset 100, last offset 199.
消费者从它停止的地方重新开始：第一次运行的最后一个偏移量是 99，第二次运行的第一个偏移量是 100。请注意，first 偏移量规范被忽略：存储的偏移量优先于偏移量规范参数。消费者在第一次运行中存储了偏移量跟踪信息，因此客户端库在第二次运行中使用它来恢复消费到正确的位置。
这结束了关于 RabbitMQ 流中消费语义的本教程。它介绍了消费者如何在流中的任何位置附加。消费应用程序可能会跟踪它们在流中到达的位置。它们可以使用本教程中演示的内置服务器端偏移量跟踪功能。它们也可以自由地使用任何其他数据存储解决方案来完成此任务。
请参阅 RabbitMQ 博客 和 流 Java 客户端文档，以获取有关偏移量跟踪的更多信息。