RabbitMQ Stream 教程 - “Hello World!”
简介
先决条件
本教程假设您已安装 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 中引入。更多信息请访问此处。
“Hello World”
(使用 Rust Stream Client)
在本教程的这一部分,我们将用 Rust 编写两个程序;一个生产者,发送单条消息;一个消费者,接收消息并打印出来。我们将略过 Rust 客户端 API 的一些细节,专注于这个非常简单的事情,以便开始入门。这是 RabbitMQ Streams 的“Hello World”。
Rust stream 客户端库
RabbitMQ 使用多种协议。本教程使用 RabbitMQ stream 协议,该协议是 RabbitMQ streams 的专用协议。RabbitMQ 有许多不同语言的客户端,请参阅每种语言的 stream 客户端库。我们将使用 RabbitMQ 提供的 Rust stream 客户端。
RabbitMQ Rust stream 客户端 0.4.2 及更高版本通过 crates.io 分发。
本教程假设您在 Windows 上使用 powershell。在 MacOS 和 Linux 上,几乎任何 shell 都可以工作。
设置
首先,让我们验证您的 Rust 工具链是否在 PATH
中
rustc --version
cargo --version
运行该命令应生成帮助消息。
本教程的可执行版本可以在 RabbitMQ 教程存储库中找到。
现在让我们创建项目
cargo new rust-stream
cd rust-stream
从 GitHub 上的 Cargo.toml 文件中复制所需的依赖项到您自己的 Cargo.toml
文件中。
现在我们已经设置了 Rust 项目,可以编写一些代码了。
发送
我们将把消息生产者(发送者)称为 send.rs
,消息消费者(接收者)称为 receive.rs
。生产者将连接到 RabbitMQ,发送单条消息,然后退出。
在 src/bin
目录中创建 send.rs
文件。
在 send.rs
中,我们需要一些 use
声明
use rabbitmq_stream_client::error::StreamCreateError;
use rabbitmq_stream_client::types::{ByteCapacity, Message, ResponseCode};
然后我们可以创建与服务器的连接
use rabbitmq_stream_client::Environment;
let environment = Environment::builder().build().await?;
stream Rust 客户端的入口点是 Environment
。它用于配置 RabbitMQ stream 发布者、stream 消费者和 stream 本身。
它抽象了套接字连接,并负责协议版本协商和身份验证等等。
本教程假设 stream 发布者和消费者连接到本地运行的 RabbitMQ 节点,即 localhost。要连接到不同机器上的节点,只需在 Environment::builder()
上指定目标主机名或 IP 地址。
接下来,让我们创建一个生产者。
生产者还将声明一个 stream,它将向该 stream 发布消息,然后发布一条消息
let stream = "hello-rust-stream";
let create_response = environment
.stream_creator()
.max_length(ByteCapacity::GB(5))
.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);
}
}
}
}
let producer = environment.producer().build(stream).await?;
producer
.send_with_confirm(Message::builder().body("Hello, World!").build())
.await?;
stream 声明操作是幂等的:stream 仅在尚不存在时才会被创建。在这种情况下,我们忽略错误 StreamAlreadyExists
,因为我们不关心 stream 是否已存在。
stream 是一个仅追加日志抽象,允许重复消费消息,直到消息过期。始终定义保留策略是一个好习惯。在上面的示例中,stream 的大小限制为 5 GiB。
消息内容是一个字节数组。应用程序可以使用任何适当的格式(例如 JSON、MessagePack 等)对需要传输的数据进行编码。
当上面的代码完成运行时,生产者连接和 stream 系统连接将被关闭。我们的生产者就完成了。
每次运行生产者时,它都会向服务器发送单条消息,并且该消息将被附加到 stream。
完整的 send.rs 文件可以在 GitHub 上找到。
发送不起作用!
如果这是您第一次使用 RabbitMQ,并且没有看到“Sent”消息,那么您可能会挠头想知道哪里出了问题。可能是 broker 启动时没有足够的可用磁盘空间(默认情况下至少需要 50 MB 可用空间),因此拒绝接受消息。检查 broker 日志文件以查看是否记录了 资源警报,并在必要时降低可用磁盘空间阈值。配置指南将向您展示如何设置
disk_free_limit
。另一个原因可能是程序在消息到达 broker 之前就退出了。在某些客户端库中,发送是异步的:函数立即返回,但消息在通过网络传输之前会排队到 IO 层中。发送程序要求用户按键以完成该过程:消息有足够的时间到达 broker。stream 协议提供了一种确认机制,以确保 broker 接收到出站消息,但为了简单起见,本教程未使用此机制。
接收
本教程的另一部分,消费者,将连接到 RabbitMQ 节点并等待消息被推送到它。与本教程中发布单条消息并停止的生产者不同,消费者将持续运行,消费 RabbitMQ 将推送给它的消息,并打印接收到的有效负载。
在 src/bin
目录中创建 receive.rs
文件。
与 send.rs
类似,receive.rs
需要 use
声明
use std::io::stdin;
use rabbitmq_stream_client::error::StreamCreateError;
use rabbitmq_stream_client::types::{ByteCapacity, OffsetSpecification, ResponseCode};
use futures::{StreamExt};
use tokio::task;
当涉及到初始设置时,消费者部分与生产者部分非常相似;我们使用默认连接设置并声明消费者将从中消费的 stream。
use rabbitmq_stream_client::Environment;
let environment = Environment::builder().build().await?;
let stream = "hello-rust-stream";
let create_response = environment
.stream_creator()
.max_length(ByteCapacity::GB(5))
.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);
}
}
}
}
请注意,消费者部分也声明了 stream。这是为了允许任何一部分首先启动,无论是生产者还是消费者。
我们使用 Consumer
结构体来创建消费者。
Consumer
提供了方法 next()
来从 stream 获取下一条消息。
offset
定义了消费者的起始点。在本例中,消费者从 stream 中可用的第一条消息开始。
let mut consumer = environment
.consumer()
.offset(OffsetSpecification::First)
.build(stream)
.await
.unwrap();
let handle = consumer.handle();
task::spawn(async move {
while let Some(delivery) = consumer.next().await {
let d = delivery.unwrap();
println!("Got message: {:#?} with offset: {}",
d.message().data().map(|data| String::from_utf8(data.to_vec()).unwrap()),
d.offset(),);
}
});
完整的 receive.rs 文件可以在 GitHub 上找到。
将它们放在一起
为了运行这两个示例,请打开两个终端(shell)选项卡。
本教程的两个部分可以按任何顺序运行,因为它们都声明了 stream。让我们首先运行消费者,以便在第一个发布者启动时,消费者将打印它
cargo run --bin receive
然后运行生产者
cargo run --bin send
消费者将打印它从 RabbitMQ 通过发布者获得的消息。消费者将保持运行,等待新的交付。尝试多次重新运行发布者以观察这一点。
Streams 与队列不同,因为它们是可以重复消费的消息的仅追加日志。当多个消费者从一个 stream 消费时,他们将从第一条可用消息开始。