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 Streams 在 RabbitMQ 3.9 中引入。更多信息请访问此处。
偏移量跟踪
设置
本教程的一部分包括用 Rust 编写两个程序；一个生产者，发送一连串的消息，并在最后发送一个标记消息；一个消费者，接收消息并在收到标记消息时停止。它展示了消费者如何浏览流，甚至可以在之前的执行中从停止的地方重新开始。
本教程使用流 Rust 客户端。请确保按照第一个教程中的设置步骤进行操作。
本教程的可执行版本可以在RabbitMQ 教程仓库中找到。
请注意，可执行版本已实现本教程末尾解释的服务器端偏移量跟踪部分，在测试此场景时需要考虑这一点。
发送程序称为 offset_tracking_send.rs，接收程序称为 receive_offset_tracking.rs。本教程重点介绍客户端库的使用，因此应使用存储库中的最终代码创建文件的脚手架（例如导入、主函数等）。
发送
发送程序首先实例化 Environment 并创建流
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 的使用：主例程在确认回调确认所有消息之前一直等待它。这确保了代理在关闭程序之前收到了所有消息。
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 实例并确保流也已创建。此代码部分与发送程序相同，因此为了简洁起见，在接下来的代码片段中跳过。
接收程序启动一个消费者，该消费者从流的开头 OffsetSpecification::First 附加。它使用两个变量：first_offset 和 last_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())
}
探索流
要运行这两个示例，请打开两个终端（shell）选项卡。
在第一个选项卡中，运行发送方以发布一系列消息
cargo run --bin send_offset_tracking
输出如下
Publishing 100 messages
Messages confirmed: True
现在让我们运行接收方。打开一个新选项卡。请记住，它应该从流的开头开始，因为使用了 OffsetSpecification::First 偏移量规范。
cargo run --bin receive_offset_tracking
输出如下
Started consuming
consuming first message
Done consuming first_offset: 0 last_offset: 99
流可以看作一个数组，其中元素是消息。偏移量是给定消息在数组中的索引。
流与队列不同：消费者可以读取和重新读取相同的消息，并且消息保留在流中。
让我们尝试使用 OffsetSpecification::Offset 规范附加到不同于 0 的给定偏移量来使用此功能。在为消费者创建环境时，将 OffsetSpecification 变量从
consumer = environment
.consumer()
.offset(OffsetSpecification::First)
.build(stream)
.await
.unwrap();
更改为
consumer = environment
.consumer()
.offset(OffsetSpecification::Offset(42))
.build(stream)
.await
.unwrap();
偏移量 42 是任意的，它可以是 0 到 99 之间的任何数字。再次运行接收方
cargo run --bin receive_offset_tracking
输出如下
Started consuming:
First message received.
Done consuming first_offset: 42 last_offset: 99
还有一种方法可以附加到流的末尾，以便在消费者创建时仅查看新消息。这是 OffsetSpecification::Next 偏移量规范。让我们试试
consumer = environment
.consumer()
.offset(OffsetSpecification::Next)
.build(stream)
.await
.unwrap();
运行接收方
cargo run --bin receive_offset_tracking
这次消费者没有收到任何消息
Started consuming
它正在等待流中的新消息。让我们通过再次运行发送方来发布一些消息。返回第一个选项卡
cargo run --bin send_offset_tracking
等待程序退出并切换回接收方选项卡。消费者收到了新消息
Started consuming
First message received.
Done consuming first_offset: 100 last_offset: 199
接收方停止是因为发送方在流的末尾放置了新的标记消息。
本节展示了如何“浏览”流：从开头、从任何偏移量，甚至用于新消息。下一节介绍如何利用服务器端偏移量跟踪来恢复消费者在先前执行中停止的位置。
服务器端偏移量跟踪
RabbitMQ Streams 提供服务器端偏移量跟踪以在流中存储给定消费者的进度。如果消费者因任何原因停止（崩溃、升级等），它将能够从之前停止的位置重新附加，以避免处理相同的消息。
RabbitMQ Streams 提供了用于偏移量跟踪的 API，但可以使用其他解决方案来存储正在使用应用程序的进度。这可能取决于用例，但关系数据库也可以是一个不错的解决方案。
让我们修改接收方以存储已处理消息的偏移量。更新的行用注释概述
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.
这里没有什么令人惊讶的：消费者从流的开头获取消息，并在到达标记消息时停止。
让我们再启动一次
cargo run --bin receive_offset_tracking
输出如下
Started consuming
First message received.
Done consuming first_offset: 100 last_offset: 199
最相关的实现是
- 消费者必须有一个名称。它是存储和检索最后存储的偏移量值的密钥。
- 每 10 条消息存储一次偏移量。对于偏移量存储频率来说，这是一个异常低的值，但这对于本教程来说是可以的。现实世界中的值更可能在数百或数千范围内。
- 在关闭消费者之前，就在获取标记消息后存储偏移量。
消费者从它停止的地方重新开始：第一次运行中的最后一个偏移量是 99，第二次运行中的第一个偏移量是 100。请注意，OffsetSpecification::First 偏移量规范被忽略：存储的偏移量优先于偏移量规范参数。消费者在第一次运行中存储了偏移量跟踪信息，因此客户端库使用它在第二次运行中恢复到正确的位置。
这结束在本教程中关于 RabbitMQ Streams 中消费语义的内容。它介绍了消费者如何在流中的任何位置附加。使用应用程序可能会跟踪它们在流中达到的位置。它们可以使用本教程中演示的内置服务器端偏移量跟踪功能。它们也可以自由地使用任何其他数据存储解决方案来完成此任务。
有关偏移量跟踪的更多信息，请参阅RabbitMQ 博客。