AMQP Rust Client
The recommended Rust client is amqp-client.rs. It’s built on tokio, speaks TLS through rustls, and recovers connections, queues and consumers when recovery is turned on. Install it with:
cargo add amqp-client
cargo add tokio --features macros,rt-multi-thread,signal
Publish and consume
The example reads the broker URL from BROKER_URL and falls back to a local LavinMQ at amqp://guest:guest@localhost.
The Cargo.toml, which names the library crate rust_amqp_client_example, lives in the example repo.
Save the connection, publish and consume code as src/lib.rs:
use amqp_client::{BasicProperties, Client, Error, QueueOpts, RecoveryConfig, SubscribeOpts};
/// Connects to BROKER_URL, and recovers the connection, queues and consumers if it's lost.
pub async fn connect() -> Result<Client, Error> {
let url =
std::env::var("BROKER_URL").unwrap_or_else(|_| "amqp://guest:guest@localhost".into());
Client::builder(&url)?
.recovery(RecoveryConfig::default())
.build()
.await
}
/// Publishes a persistent message and returns once the broker has confirmed it.
pub async fn publish(client: &Client, queue: &str, message: &str) -> Result<(), Error> {
client.queue(queue, QueueOpts::default()).await?;
let props = BasicProperties::builder()
.content_type("text/plain")
.persistent()
.build()?;
client
.publish_confirmed_with_props("", queue, &message.as_bytes().to_vec(), props)
.await?;
println!(" [x] Sent '{message}'");
Ok(())
}
/// Calls `handler` for each message, with a prefetch of 10, and acks the message after it returns.
pub async fn consume(
client: &Client,
queue: &str,
mut handler: impl FnMut(String),
) -> Result<(), Error> {
let queue = client.queue(queue, QueueOpts::default()).await?;
let mut consumer = queue
.subscribe::<Vec<u8>>(SubscribeOpts {
prefetch: 10,
..Default::default()
})
.await?;
while let Some(delivery) = consumer.next().await {
match delivery {
Ok(d) => {
handler(String::from_utf8_lossy(&d.body).into_owned());
d.ack()?;
}
// Not decodable: dead-letter it instead of redelivering it forever
Err(f) => f.reject(false)?,
}
}
Ok(())
}
Save the publisher as src/bin/publisher.rs:
use rust_amqp_client_example::{connect, publish};
#[tokio::main]
async fn main() -> Result<(), amqp_client::Error> {
let client = connect().await?;
publish(&client, "hello", "Hello world!").await
}
Save the consumer as src/bin/worker.rs:
use rust_amqp_client_example::{connect, consume};
#[tokio::main]
async fn main() -> Result<(), amqp_client::Error> {
let client = connect().await?;
println!(" [*] Waiting for messages. To exit press CTRL+C");
tokio::select! {
result = consume(&client, "hello", |message| println!(" [x] Received '{message}'")) => result,
_ = tokio::signal::ctrl_c() => Ok(()),
}
}
Run cargo run --bin worker in one terminal and cargo run --bin publisher in another.
The full, runnable example lives in rust/amqp-client in cloudamqp/examples, and CI tests it against LavinMQ.
Good to know
- Recovery is opt-in. Without
.recovery(...)on the builder, a lost connection is final:publishreturns an error and consumers end. The example turns it on, so the client retries every 250 ms, declares its queues again and resumes consumers. A message can arrive twice after a reconnect, so keep handlers idempotent. - Settle every delivery. The worker acks after the handler returns. A delivery dropped without
ack,nackorrejectstays unacked until its channel closes and holds a prefetch slot until then. - Set a prefetch limit.
subscribedefaults to a prefetch of 20. The example lowers it to 10 withSubscribeOpts. - Use
amqps://for TLS. The client verifies the server certificate and hostname against the bundled Mozilla root certificates, not the system trust store.
The amqp-client.rs README covers codecs, exchanges and the low-level API.
Other Rust clients
- lapin, a widely used futures-based client
Ready to take the next steps?
Managed LavinMQ instance via CloudAMQP
LavinMQ has been built with performance and ease of use in mind - we've benchmarked a throughput of about 1,000,000 messages/sec. You can try LavinMQ without any installation hassle by creating a free instance on CloudAMQP. Signing up is a breeze.
Get started with CloudAMQP ->
Help and feedback
We welcome your feedback and are eager to address any questions you may have about this piece or using LavinMQ. Join our Slack channel to connect with us directly. You can also find LavinMQ on GitHub.