跳到主要内容

RabbitMQ Stream 教程 - Offset 跟踪

简介​

信息

前提条件​

本教程假设您已安装 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 中引入。更多信息请见 此处。

Offset 跟踪​

设置​

本教程的这一部分包括编写两个 Javascript 程序：一个生产者，它发送一连串消息，并在末尾添加一个标记消息；以及一个消费者，它接收消息并在收到标记消息时停止。它展示了消费者如何在 stream 中导航，甚至可以在之前的执行中断的地方重新开始。

本教程使用 stream Javascript 客户端。请务必按照第一个教程中的 设置步骤 进行操作。

本教程的可执行版本可以在 RabbitMQ tutorials 仓库 中找到。发送程序名为 offset_tracking_send.js，接收程序名为 offset_tracking_receive.js。本教程侧重于客户端库的用法，因此仓库中的最终代码应用于创建文件的框架（例如，导入、主函数等）。

发送​

在 offset_tracking_send.js 中，我们需要添加客户端

const rabbit = require("rabbitmq-stream-js-client");

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

const client = await rabbit.connect({
  hostname: "localhost",
  port: 5552,
  username: "guest",
  password: "guest",
  vhost: "/",
});

console.log("Making sure the stream exists...");
const streamSizeRetention = 5 * 1e9;
const streamName = "stream-offset-tracking-javascript";
await client.createStream({ stream: streamName, arguments: { "max-length-bytes": streamSizeRetention } });

然后程序创建一个 producer 并发布 100 条消息。最后一条消息的主体值设置为 marker；这是 consumer 停止消费的标记。

console.log("Creating the publisher...");
const publisher = await client.declarePublisher({ stream: streamName });

const messageCount = 100;
console.log(`Publishing ${messageCount} messages`);
for (let i = 0; i < messageCount; i++) {
  const body = i === messageCount - 1 ? "marker" : `hello ${i}`;
  await publisher.send(Buffer.from(body));
}

如果您在发送消息时遇到任何问题，我建议您查看 这里 的一些解决方案

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

接收​

接收程序 offset_tracking_receive.js 也添加了客户端，创建了到服务器的连接和一个 stream。这部分代码与发送程序中的代码相同，因此为了简洁起见，在接下来的代码片段中跳过。

接收程序启动一个消费者，该消费者从 stream 的开头 (rabbit.Offset.first()) 开始附加。它使用变量在程序结束时输出第一个和最后一个接收到的消息的 offset。

当消费者收到标记消息时停止：它将 offset 分配给一个变量，并将消息 offset 存储在服务器上。与发送者一样，当消费者完成其工作时，通道会告诉程序继续进行。

const startFrom = rabbit.Offset.first();
let firstOffset = startFrom.value;
let lastOffset = startFrom.value;
let messageCount = 0;
const consumer = await client.declareConsumer({ stream: streamName, offset: startFrom }, (message) => {
  messageCount++;
  if (messageCount === 1) {
    console.log("First message received");
    firstOffset = message.offset;
  }
  if (message.content.toString() === "marker") {
    console.log("Marker found");
    lastOffset = message.offset;
    console.log(`Done consuming, first offset was ${firstOffset}, last offset was ${lastOffset}`);
  }
});

console.log(`Start consuming...`);
await sleep(2000);

探索 Stream​

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

在第一个标签页中，运行发送者以发布一连串消息

npm run offset-tracking-publish

输出如下

Connecting...
Making sure the stream exists...
Creating the publisher...
Publishing 100 messages
Closing the connection...
done!

现在让我们运行接收者。打开一个新的标签页。请记住，由于 first offset 规范，它应该从 stream 的开头开始。

npm run offset-tracking-receive

这是输出

Connecting...
First message received
Start consuming...
Marker found
Done consuming, first offset was 0, last offset was 99
什么是 offset？

Stream 可以看作是一个数组，其中的元素是消息。Offset 是数组中给定消息的索引。

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

让我们尝试使用 offset(bigint) 规范在给定的 offset 处附加来体验此功能。将 startFrom 变量从 rabbit.Offset.first() 设置为 rabbit.Offset.offset(42)

const startFrom = rabbit.Offset.offset(42n);

Offset 42 是任意的，它可以是 0 到 99 之间的任何数字。再次运行接收者

npm run offset-tracking-receive

输出如下

Connecting...
Start consuming...
First message received
Marker found
Done consuming, first offset was 42, last offset was 99

还有一种方法可以在 stream 的末尾附加，以便仅查看消费者创建时的新消息。这是 next offset 规范。让我们尝试一下

const startFrom = rabbit.Offset.next();

运行接收者

npm run offset-tracking-receive

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

Connecting...
Start consuming...

它正在等待 stream 中的新消息。让我们再次运行发送者来发布一些消息。回到第一个标签页

npm run offset-tracking-publish

等待程序退出并切换回接收者标签页。消费者收到了新消息

First message received
Marker found
Done consuming, first offset was 100, last offset was 199

接收者停止是因为发送者在 stream 末尾放置了新的标记消息。

本节展示了如何“浏览” stream：从开头、从任何 offset 甚至对于新消息。下一节将介绍如何利用服务器端 offset 跟踪来恢复消费者在之前的执行中中断的地方。

服务器端 Offset 跟踪​

RabbitMQ Streams 提供服务器端 offset 跟踪，以存储给定消费者在 stream 中的进度。如果消费者因任何原因（崩溃、升级等）停止，它将能够重新附加到之前停止的位置，以避免处理相同的消息。

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

让我们修改接收者以存储已处理消息的 offset。更新的行用注释标出

// start consuming at the beginning of the stream
const consumerRef = "offset-tracking-tutorial"; // the consumer must a have name
let firstOffset = undefined;
let offsetSpecification = rabbit.Offset.first();
try {
  const offset = await client.queryOffset({ reference: consumerRef, stream: streamName }); // take the offset stored on the server if it exists
  offsetSpecification = rabbit.Offset.offset(offset + 1n); // start from the message after 'marker'
} catch (e) {}

let lastOffset = offsetSpecification.value;
let messageCount = 0;
const consumer = await client.declareConsumer(
  { stream: streamName, offset: offsetSpecification, consumerRef },
  async (message) => {
    messageCount++;
    if (!firstOffset && messageCount === 1) {
      firstOffset = message.offset;
      console.log("First message received");
    }
    if (messageCount % 10 === 0) {
      await consumer.storeOffset(message.offset); // store offset every 10 messages
    }
    if (message.content.toString() === "marker") {
      console.log("Marker found");
      lastOffset = message.offset;
      await consumer.storeOffset(message.offset); // store the offset on consumer closing
      await consumer.close(true);
    }
  }
);

console.log(`Start consuming...`);
await sleep(2000);
console.log(`Done consuming, first offset was ${firstOffset}, last offset was ${lastOffset}`);
process.exit(0);

最相关的更改是

  • 消费者在第一次启动时使用 rabbit.Offset.first() 从 stream 的开头附加。
  • 消费者必须有一个名称。它是存储和检索上次存储的 offset 值的键。
  • Offset 每 10 条消息存储一次。对于 offset 存储频率来说，这是一个异常低的值，但这对于本教程来说是可以的。实际应用中的值通常在数百或数千。
  • Offset 在关闭消费者之前存储，就在收到标记消息之后。

让我们运行更新后的接收者

npm run offset-tracking-receive

这是输出

Connecting...
Start consuming...
First message received
Marker found
Done consuming, first offset was 0, last offset was 99

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

让我们发布另一批 100 条消息，并再次启动接收者

npm run offset-tracking-publish
npm run offset-tracking-receive

这是输出

Connecting...
Start consuming...
Marker found
Done consuming, first offset was 100, last offset was 201

消费者从上次停止的地方重新启动：第一次运行的最后一个 offset 是 99，第二次运行的第一个 offset 是 100。请注意第二次运行，offset 规范是通过 queryOffset 方法获取的。消费者在第一次运行中存储了 offset 跟踪信息，因此客户端库提供了检索它的可能性，以便在第二次运行中在正确的位置恢复消费。

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

有关 offset 跟踪的更多信息，请参阅 RabbitMQ 博客 和 stream Javascript nodejs 客户端文档。

© . All rights reserved.