跳至主内容

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

偏移量跟踪​

设置​

本教程的这一部分将编写两个 Java 程序：一个生产者，它发送一批带有标记消息的消息，以及一个消费者，它接收消息并在收到标记消息时停止。它展示了消费者如何遍历消息流，甚至可以从之前的执行中断处继续。

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

本教程的可执行版本可以在 RabbitMQ 教程仓库 中找到。发送程序名为 OffsetTrackingSend.java，接收程序名为 OffsetTrackingReceive.java。本教程侧重于客户端库的使用，因此应使用仓库中的最终代码来创建文件的脚手架（例如，导入、main 函数等）。

发送​

发送程序首先实例化 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 Streams 提供服务器端偏移量跟踪，用于存储流中给定消费者的进度。如果消费者因任何原因停止（崩溃、升级等），它将能够从先前停止的位置重新连接，以避免处理相同的消息。

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

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

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