跳到主要内容

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 追踪​

设置​

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

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

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

请注意，可执行版本已经实现了本教程末尾解释的服务器端 Offset 追踪部分，在测试此场景时需要考虑这一点。

发送程序名为 offset_tracking_send.rs，接收程序名为 receive_offset_tracking.rs。本教程侧重于客户端库的使用，因此应使用仓库中的最终代码来创建文件的框架（例如，导入、main 函数等）。

发送​

发送程序首先实例化 Environment 并创建 stream

let create_response = environment
    .stream_creator()
    .max_length(ByteCapacity::GB(2))
    .create(stream)
    .await;

if let Err(e) = create_response {
    if let StreamCreateError::Create { stream, status } = e {
        match status {
            // we can ignore this error because the stream already exists
            ResponseCode::StreamAlreadyExists => {}
            err => {
                println!("Error creating stream: {:?} {:?}", stream, err);
                std::process::exit(1);
            }
        }
    }
}

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

请注意 tokio::sync::Notify 的使用：主例程正在等待它，直到所有消息都通过确认回调确认。这确保了 broker 在程序关闭之前收到所有消息。

let producer = environment.producer().build(stream).await?;

println!("Publishing {:?} messages", message_count);

for i in 0..message_count {
    let msg;
    if i < message_count - 1 {
        msg = Message::builder().body(format!("hello{}", i)).build();
    } else {
        msg = Message::builder().body(format!("marker{}", i)).build();
    };

    let counter = confirmed_messages.clone();
    let notifier = notify_on_send.clone();
    producer
        .send(msg, move |_| {
            let inner_counter = counter.clone();
            let inner_notifier = notifier.clone();
            async move {
                if inner_counter.fetch_add(1, Ordering::Relaxed) == message_count - 1 {
                    inner_notifier.notify_one();
                }
            }
        })
        .await?;
}

notify_on_send.notified().await;
println!("Messages confirmed: True");
producer.close().await?;

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

接收​

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

接收程序启动一个消费者，该消费者从 stream 的开头 OffsetSpecification::First 处附加。它使用两个变量：first_offset 和 last_offset，以便在程序结束时输出第一个和最后一个接收到的消息的 offset。

当消费者收到标记消息时停止：它将 offset 分配给 last_offset 变量并关闭消费者。

let mut first_offset: Option<u64> = None;
let mut last_offset: Option<u64> = None;
let mut consumer = environment
    .consumer()
    .offset(OffsetSpecification::First)
    .build(stream)
    .await
    .unwrap();

while let Some(delivery) = consumer.next().await {
    let d = delivery.unwrap();

    if !first_offset.is_some()  {
        println!("First message received");
        first_offset = Some(d.offset());
    }

    if  String::from_utf8_lossy(d.message().data().unwrap()).contains("marker")
    {
        last_offset = Some(d.offset());
        let handle = consumer.handle();
        _ = handle.close().await;
        break;
    }
}

if first_offset.is_some() {
    println!(
        "Done consuming first_offset: {:?} last_offset: {:?}  ", first_offset.unwrap(), last_offset.unwrap())
}

探索 Stream​

为了运行这两个示例，请打开两个终端（shell）选项卡。

在第一个选项卡中，运行发送器以发布一连串消息

 cargo run --bin send_offset_tracking

输出如下

Publishing 100 messages
Messages confirmed: True

现在让我们运行接收器。打开一个新的选项卡。请记住，由于 OffsetSpecification::First offset 规范，它应该从 stream 的开头开始。

  cargo run --bin receive_offset_tracking

这是输出

Started consuming
consuming first message
Done consuming first_offset: 0 last_offset: 99
什么是 offset？

可以将 stream 视为一个数组，其中的元素是消息。Offset 是数组中给定消息的索引。

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

让我们尝试使用 OffsetSpecification::Offset 规范从不同于 0 的给定 offset 处附加此功能。在为 Consumer 创建环境时，将 OffsetSpecification 变量从

consumer = environment
    .consumer()
    .offset(OffsetSpecification::First)
    .build(stream)
    .await
    .unwrap();

改为

consumer = environment
    .consumer()
    .offset(OffsetSpecification::Offset(42))
    .build(stream)
    .await
    .unwrap();

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

 cargo run --bin receive_offset_tracking

这是输出

Started consuming:
First message received.
Done consuming first_offset: 42 last_offset: 99

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

consumer = environment
    .consumer()
    .offset(OffsetSpecification::Next)
    .build(stream)
    .await
    .unwrap();

运行接收器

 cargo run --bin receive_offset_tracking

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

Started consuming

它正在等待 stream 中的新消息。让我们通过再次运行发送器来发布一些消息。回到第一个选项卡

 cargo run --bin send_offset_tracking

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

Started consuming
First message received.
Done consuming first_offset: 100 last_offset: 199

接收器因发送器放在 stream 末尾的新标记消息而停止。

本节介绍了如何“浏览” stream：从头开始，从任何 offset 开始，甚至只看新消息。下一节介绍如何利用服务器端 offset 追踪来恢复消费者在上次执行中停止的位置。

服务器端 Offset 追踪​

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

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

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

let mut first_offset: Option<u64> = None;
let mut last_offset: Option<u64> = None;
let mut consumer = environment
    .consumer()
    // The consumer needs a name to use Server-Side Offset Tracking
    .name("consumer-1")
    .offset(OffsetSpecification::First)
    .build(stream)
    .await
    .unwrap();

println!("Started consuming");

// We can query if a stored offset exists
let mut stored_offset: u64 = consumer.query_offset().await.unwrap_or_else(|_| 0);

if stored_offset >  0 {
    stored_offset += 1;
}
consumer = environment
    .consumer()
    // The consumer needs a name to use Server-Side Offset Tracking
    .name("consumer-1")
    .offset(OffsetSpecification::Offset(stored_offset))
    .build(stream)
    .await
    .unwrap();

let mut received_messages: i64 = -1;
while let Some(delivery) = consumer.next().await {
    let d = delivery.unwrap();

    if !first_offset.is_some()  {
        first_offset = Some(d.offset());
    }
    consumer = environment
        .consumer()
        // The consumer needs a name to use Server-Side Offset Tracking
        .name("consumer-1")
        .offset(OffsetSpecification::Offset(stored_offset))
        .build(stream)
        .await
        .unwrap();

    let mut received_messages: i64 = -1;
    while let Some(delivery) = consumer.next().await {
        let d = delivery.unwrap();

        if !first_offset.is_some()  {
            println!("First message received");
            first_offset = Some(d.offset());
        }
        received_messages = received_messages + 1;
        if received_messages % 10 == 0
            || String::from_utf8_lossy(d.message().data().unwrap()).contains("marker")
        {
            // We store the offset in the server
            let _ = consumer
                .store_offset(d.offset())
                .await
                .unwrap_or_else(|e| println!("Err: {}", e));
            if String::from_utf8_lossy(d.message().data().unwrap()).contains("marker") {
                last_offset = Some(d.offset());
                let handle = consumer.handle();
                _ = handle.close().await;
                break;
            }
        }
    }
}

if first_offset.is_some() {
    println!(
        "Done consuming first_offset: {:?} last_offset: {:?}  ", first_offset.unwrap(), last_offset.unwrap())
}

让我们运行接收器

 cargo run --bin receive_offset_tracking

这是输出

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

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

让我们再次启动它

cargo run --bin receive_offset_tracking

这是输出

Started consuming
First message received.
Done consuming first_offset: 100 last_offset: 199

最相关的实现是

  • 消费者必须有一个名称。它是存储和检索上次存储的 offset 值的键。
  • 每 10 条消息存储一次 offset。对于 offset 存储频率来说，这是一个异常低的值，但这对于本教程来说是可以的。现实世界中的值通常在数百或数千。
  • Offset 在关闭消费者之前存储，就在获取标记消息之后。

消费者从它上次停止的地方准确地重新启动：第一次运行的最后一个 offset 是 99，第二次运行的第一个 offset 是 100。请注意，OffsetSpecification::First offset 规范被忽略：存储的 offset 优先于 offset 规范参数。消费者在第一次运行中存储了 offset 追踪信息，因此客户端库使用它在第二次运行中从正确的位置恢复消费。

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

有关 offset 追踪的更多信息，请参阅RabbitMQ 博客。

© . All rights reserved.