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

偏移量跟踪

设置

本教程的这一部分包括编写两个 Python 程序：一个生产者，它发送一系列带有标记消息的消息，以及一个消费者，它接收消息并在收到标记消息时停止。它演示了消费者如何浏览流，甚至可以从之前的执行中断点恢复。

本教程使用了 rstream Python 客户端。请确保遵循第一个教程中的 设置步骤。

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

请注意，可执行版本已经实现了本教程结尾处解释的 服务器端偏移跟踪 功能，在测试此场景时需要将这一点考虑在内。

发送程序名为 offset_tracking_send.py，接收程序名为 offset_tracking_receive.py。本教程重点介绍客户端库的用法，因此应使用仓库中的最终代码来创建文件的脚手架（例如，导入、主函数等）。

发送

发送程序创建一个 Producer 实例并发布 100 条消息。

最后一条消息的正文设置为 marker；这是消费者停止消费的标记。

程序通过 _on_publish_confirm_client 回调来处理消息确认。

请注意 asyncio.Condition 的使用：主例程会一直等待它，直到在 _on_publish_confirm_client 回调中收到所有消息的确认，然后该回调会通知主例程。这确保了代理在程序关闭之前收到了所有消息。

STREAM = "stream-offset-tracking-python"
MESSAGES = 100
# 2GB
STREAM_RETENTION = 2000000000
confirmed_messages = 0
all_confirmed_messages_cond = asyncio.Condition()

async def _on_publish_confirm_client(confirmation: ConfirmationStatus) -> None:
    global confirmed_messages
    if confirmation.is_confirmed:
        confirmed_messages = confirmed_messages + 1
        if confirmed_messages == 100:
            async with all_confirmed_messages_cond:
                all_confirmed_messages_cond.notify()

async def publish():
    async with Producer("localhost", username="guest", password="guest") as producer:
        # create a stream if it doesn't already exist
        await producer.create_stream(
            STREAM, exists_ok=True, arguments={"max-length-bytes": STREAM_RETENTION}
        )

        print("Publishing {} messages".format(MESSAGES))
        # Send 99 hello message
        for i in range(MESSAGES - 1):
            amqp_message = AMQPMessage(
                body=bytes("hello: {}".format(i), "utf-8"),
            )

            await producer.send(
                stream=STREAM,
                message=amqp_message,
                on_publish_confirm=_on_publish_confirm_client,
            )
        # Send a final marker message
        amqp_message = AMQPMessage(
            body=bytes("marker: {}".format(i + 1), "utf-8"),
        )

        await producer.send(
            stream=STREAM,
            message=amqp_message,
            on_publish_confirm=_on_publish_confirm_client,
        )

        async with all_confirmed_messages_cond:
            await all_confirmed_messages_cond.wait()

        print("Messages confirmed.")


asyncio.run(publish())

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

接收

接收程序启动一个消费者，该消费者附加到流的开头 ConsumerOffsetSpecification(OffsetType.FIRST)。它使用两个变量：first_offset 和 last_offset，用于在程序结束时输出接收到的第一条和最后一条消息的偏移量。on_message 回调处理传入的消息。当接收到标记消息时，消费者停止：它将消息偏移量分配给 last_offset 变量并关闭消费者。

message_count = -1
first_offset = -1
last_offset = -1
STREAM_NAME = "stream-offset-tracking-python"
# 2GB
STREAM_RETENTION = 2000000000

async def on_message(msg: AMQPMessage, message_context: MessageContext):
    global first_offset
    global last_offset

    offset = message_context.offset
    if first_offset == -1:
        print("First message received")
        first_offset = offset

    consumer = message_context.consumer
    stream = message_context.consumer.get_stream(message_context.subscriber_name)

    if "marker" in str(msg):
        last_offset = offset
        await consumer.close()

async def consume():

    global first_offset
    global last_offset

    consumer = Consumer(
        host="localhost",
        port=5552,
        username="guest",
        password="guest",
    )

    await consumer.create_stream(
        STREAM_NAME, exists_ok=True, arguments={"max-length-bytes": STREAM_RETENTION}
    )

    try:
        await consumer.start()
        print("Starting consuming Press control +C to close")

        await consumer.subscribe(
            stream=STREAM_NAME,
            callback=on_message,
            decoder=amqp_decoder,
            offset_specification=ConsumerOffsetSpecification(
                OffsetType.FIRST
            ),
        )
        await consumer.run()

    except (KeyboardInterrupt, asyncio.exceptions.CancelledError):
        await consumer.close()

    # give time to the consumer task to close the consumer
    await asyncio.sleep(1)

    if first_offset != -1:
        print(
            "Done consuming first_offset: {} last_offset {} ".format(
                first_offset, last_offset
            )
        )


with asyncio.Runner() as runner:
    runner.run(consume())

探索流

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

在第一个选项卡中，运行发送程序以发布一系列消息。

 python3 offset_tracking_send.py

输出如下：

Publishing 100 messages...
Messages confirmed: true.

现在我们来运行接收器。打开一个新标签页。记住，由于 FIRST 偏移量规范，它应该从流的开头开始。

 python3 offset_tracking_receive.py

这是输出：

Started consuming: Press control +C to close
First message received.
Done consuming, first offset 0, last offset 99.

什么是偏移量？

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

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

让我们通过使用 ConsumerOffsetSpecification(OffsetType.OFFSET, long) 规范来尝试此功能，以附加到与 0 不同的给定偏移量。在消费者的 subscribe 方法中，将 ConsumerOffsetSpecification 变量从

offset_specification=ConsumerOffsetSpecification(
    OffsetType.FIRST
),

更改为

offset_specification = ConsumerOffsetSpecification(
    OffsetType.OFFSET, 42
)

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

 python3 offset_tracking_receive.py

输出如下：

Started consuming: Press control +C to close
First message received.
Done consuming, first offset 42, last offset 99.

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

offset_specification = ConsumerOffsetSpecification(
    OffsetType.NEXT)

运行接收程序。

 python3 offset_tracking_receive.py

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

Started consuming: Press control +C to close

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

 python3 offset_tracking_send.py

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

Started consuming: Press control +C to close
First message received.
Done consuming, first offset 100, last offset 199.

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

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

服务器端偏移量跟踪

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

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

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

async def on_message(msg: AMQPMessage, message_context: MessageContext):
    # variable to keep track of the number of received messages
    global message_count
    global first_offset
    global last_offset

    offset = message_context.offset
    if first_offset == -1:
        print("First message received")
        first_offset = offset

    consumer = message_context.consumer
    stream = message_context.consumer.get_stream(message_context.subscriber_name)

    # store the offset after every 10 messages received
    message_count = message_count + 1

    if message_count % 10 == 0:
        # store_message needs to take a subscriber_name parameter
        await consumer.store_offset(
            stream=stream,
            offset=offset,
            subscriber_name=message_context.subscriber_name,
        )

    # store the offset after receiving the marker message
    if "marker" in str(msg):
        await consumer.store_offset(
            stream=stream,
            offset=offset,
            subscriber_name=message_context.subscriber_name,
        )
        last_offset = offset
        await consumer.close()

async def consume():
    # the offset to start consuming from
    stored_offset = -1
    global first_offset
    global last_offset

    # start a consumer and creates the stream is not exist (same as before...)

    try:
        await consumer.start()
        print("Started consuming: Press control +C to close")
        try:
            # query_offset must take a subscriber_name as parameter
            stored_offset = await consumer.query_offset(
                stream=STREAM_NAME, subscriber_name="subscriber_1"
            )
        except OffsetNotFound as offset_exception:
            print(f"Offset not previously stored. {offset_exception}")

        except ServerError as server_error:
            print(f"Server error: {server_error}")
            exit(1)

        # if no offset was previously stored start from the first offset
        stored_offset = stored_offset + 1
        await consumer.subscribe(
            stream=STREAM_NAME,
             # We explicitely need to assign a name to the consumer
            subscriber_name="subscriber_1",       
            callback=on_message,
            decoder=amqp_decoder,
            offset_specification=ConsumerOffsetSpecification(
                OffsetType.OFFSET, stored_offset
            ),
        )
        await consumer.run()

    except (KeyboardInterrupt, asyncio.exceptions.CancelledError):
        await consumer.close()

最重要的更改是

• 消费者必须有一个名称。它是存储和检索最后一个存储的偏移量值的关键。
• 偏移量每 10 条消息存储一次。对于偏移量存储频率而言，这是一个异常低的值，但对于本教程来说没问题。实际世界中的值通常是几百或几千。
• 偏移量在关闭消费者之前存储，就在收到标记消息之后。

现在运行接收器

 python3 offset_tracking_receive.py

这是输出：

Started consuming: Press control +C to close
First message received.
Done consuming, first offset 0, last offset 99.

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

让我们再次启动它。

 python3 offset_tracking_receive.py

这是输出：

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

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

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

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