RabbitMQ and NATS

RabbitMQ routes messages to queues through exchanges. NATS routes messages by subject and adds persistence through JetStream. This lesson uses lapin and async-nats with the Storefront order events from messaging and Kafka.

Core terms

AMQP model (RabbitMQ)

Exchange: receives published messages and routes copies to queues. The producer never names a queue.
Queue: stores messages until a consumer acknowledges them.
Binding: a rule that links an exchange to a queue, with a pattern.
Exchange types: direct (exact routing key), fanout (all bound queues), topic (pattern match, * is one word and # is zero or more words).
Ack: the consumer confirms a message and the broker deletes it. An unacked message returns to the queue when the channel closes.
Prefetch: the number of unacked messages a consumer may hold. It bounds memory and spreads work.
Dead-letter exchange (DLX): receives messages that a consumer rejects without requeue.
Durability: a durable queue and a persistent message (delivery mode 2) survive a broker restart. Publisher confirms tell the producer the broker took the message.

NATS model

Subject: a dotted name such as orders.created. Wildcards: * is one token, > is the rest.
Core NATS: at-most-once delivery. Nothing is stored, and a subscriber that is offline misses the message.
Queue group: subscribers in one group share messages, so each message goes to one member.
Request-reply: the requester publishes with a reply subject and waits for one response.
JetStream: a persistence layer. A stream stores messages for subjects. A durable consumer tracks progress, needs acks, and gets redelivery after the ack wait. Replay starts from the beginning, a sequence number, or a time.

Project setup

Terminal

docker run -d --name rabbit -p 5672:5672 -p 15672:15672 rabbitmq:4-management
docker run -d --name nats -p 4222:4222 nats:2.10 -js

Cargo.toml

TOML

[package]
name = "storefront-brokers"
version = "0.1.0"
edition = "2021"

[dependencies]
axum = "0.8"
lapin = "2.5"
tokio-executor-trait = "2.1"
tokio-reactor-trait = "1.1"
async-nats = "0.38"
tokio = { version = "1", features = ["full"] }
futures-util = "0.3"
serde = { version = "1", features = ["derive"] }
serde_json = "1"

Example 1: RabbitMQ topology and a confirmed publisher

The topology is declared in one shared file so every program creates the same exchanges and queues. fulfillment receives order.* (one word after order.). audit receives order.#. Rejected messages from fulfillment go to orders.dead through orders.dlx.

src/lib.rs

Rust

use lapin::{
    options::{
        ExchangeDeclareOptions, QueueBindOptions, QueueDeclareOptions,
    },
    types::{AMQPValue, FieldTable},
    Channel, Connection, ConnectionProperties, ExchangeKind,
};

pub const URI: &str = "amqp://guest:guest@127.0.0.1:5672/%2f";

fn durable_exchange() -> ExchangeDeclareOptions {
    ExchangeDeclareOptions {
        durable: true,
        ..Default::default()
    }
}

fn durable_queue() -> QueueDeclareOptions {
    QueueDeclareOptions {
        durable: true,
        ..Default::default()
    }
}

pub async fn connect() -> lapin::Result<Connection> {
    let properties = ConnectionProperties::default()
        .with_executor(tokio_executor_trait::Tokio::current())
        .with_reactor(tokio_reactor_trait::Tokio);
    Connection::connect(URI, properties).await
}

pub async fn declare(channel: &Channel) -> lapin::Result<()> {
    channel
        .exchange_declare("orders", ExchangeKind::Topic, durable_exchange(), FieldTable::default())
        .await?;
    channel
        .exchange_declare("orders.dlx", ExchangeKind::Fanout, durable_exchange(), FieldTable::default())
        .await?;

    channel
        .queue_declare("orders.dead", durable_queue(), FieldTable::default())
        .await?;
    channel
        .queue_bind("orders.dead", "orders.dlx", "", QueueBindOptions::default(), FieldTable::default())
        .await?;

    let mut arguments = FieldTable::default();
    arguments.insert(
        "x-dead-letter-exchange".into(),
        AMQPValue::LongString("orders.dlx".into()),
    );
    channel
        .queue_declare("fulfillment", durable_queue(), arguments)
        .await?;
    channel
        .queue_bind("fulfillment", "orders", "order.*", QueueBindOptions::default(), FieldTable::default())
        .await?;

    channel
        .queue_declare("audit", durable_queue(), FieldTable::default())
        .await?;
    channel
        .queue_bind("audit", "orders", "order.#", QueueBindOptions::default(), FieldTable::default())
        .await?;
    Ok(())
}

src/bin/rabbit_publisher.rs

Rust

use lapin::{
    options::{BasicPublishOptions, ConfirmSelectOptions},
    publisher_confirm::Confirmation,
    BasicProperties,
};
use storefront_brokers::{connect, declare};

#[tokio::main]
async fn main() -> lapin::Result<()> {
    let connection = connect().await?;
    let channel = connection.create_channel().await?;
    declare(&channel).await?;
    channel.confirm_select(ConfirmSelectOptions::default()).await?;

    for key in ["order.created", "order.paid"] {
        let confirm = channel
            .basic_publish(
                "orders",
                key,
                BasicPublishOptions::default(),
                br#"{"order_id":7}"#,
                BasicProperties::default().with_delivery_mode(2),
            )
            .await?
            .await?;
        println!("{key} confirmed: {}", matches!(confirm, Confirmation::Ack(_)));
    }
    Ok(())
}

basic_publish returns a future for the broker confirmation. The second .await waits for it, and skipping it hides a lost message.
The .with_delivery_mode(2) call marks the message persistent.

Output

order.created confirmed: true
order.paid confirmed: true

Each queue now holds both messages.

Example 2: A consumer with acks, prefetch, and dead-lettering

The worker holds at most 10 unacked messages. It acks valid messages. It rejects invalid ones with requeue: false, which sends them to the DLX. Requeuing a message that can never succeed would loop forever.

src/bin/rabbit_worker.rs

Rust

use futures_util::StreamExt;
use lapin::{
    options::{BasicAckOptions, BasicConsumeOptions, BasicNackOptions, BasicQosOptions},
    types::FieldTable,
};
use storefront_brokers::{connect, declare};

#[tokio::main]
async fn main() -> lapin::Result<()> {
    let connection = connect().await?;
    let channel = connection.create_channel().await?;
    declare(&channel).await?;
    channel.basic_qos(10, BasicQosOptions::default()).await?;

    let mut consumer = channel
        .basic_consume(
            "fulfillment",
            "worker-1",
            BasicConsumeOptions::default(),
            FieldTable::default(),
        )
        .await?;

    while let Some(delivery) = consumer.next().await {
        let delivery = delivery?;
        match serde_json::from_slice::<serde_json::Value>(&delivery.data) {
            Ok(value) => {
                println!("handled {} {}", delivery.routing_key, value["order_id"]);
                delivery.ack(BasicAckOptions::default()).await?;
            }
            Err(_) => {
                println!("rejected {}", delivery.routing_key);
                delivery
                    .nack(BasicNackOptions {
                        requeue: false,
                        ..Default::default()
                    })
                    .await?;
            }
        }
    }
    Ok(())
}

With the two messages from Example 1 in the queue, the worker prints (stop it with Ctrl-C):

Output

handled order.created 7
handled order.paid 7

Publishing a body that is not JSON prints rejected order.created and the message appears in orders.dead. Start several copies of the worker and the broker spreads messages between them, at most 10 unacked each.

Example 3: NATS subjects and request-reply

Subscriptions and publishes are cheap and need no declaration. This program subscribes to orders.*, publishes two messages, and answers a request on inventory.check.

src/bin/nats_core.rs

Rust

use futures_util::StreamExt;

#[tokio::main]
async fn main() -> Result<(), async_nats::Error> {
    let client = async_nats::connect("127.0.0.1:4222").await?;

    let mut orders = client.subscribe("orders.*").await?;
    let mut requests = client.subscribe("inventory.check").await?;

    let responder = client.clone();
    tokio::spawn(async move {
        while let Some(request) = requests.next().await {
            if let Some(reply) = request.reply {
                responder.publish(reply, "in_stock".into()).await.unwrap();
            }
        }
    });

    client.publish("orders.created", "order 7".into()).await?;
    client.publish("orders.paid", "order 7".into()).await?;
    for _ in 0..2 {
        let message = orders.next().await.unwrap();
        println!("{} -> {}", message.subject, String::from_utf8_lossy(&message.payload));
    }

    let reply = client.request("inventory.check", "sku-1".into()).await?;
    println!("reply: {}", String::from_utf8_lossy(&reply.payload));
    Ok(())
}

Output

orders.created -> order 7
orders.paid -> order 7
reply: in_stock

Replacing subscribe with queue_subscribe(subject, group) makes subscribers with the same group name share messages.
A subscriber that is not connected when a message is published never sees it. JetStream fixes this.

Example 4: JetStream persistence

A stream stores every message published to its subjects. A durable pull consumer asks for batches, so a slow consumer controls its own intake. publish returns two futures: the first sends, and the second resolves to the storage acknowledgment.

src/bin/nats_jetstream.rs

Rust

use async_nats::jetstream::{self, consumer::pull, stream};
use futures_util::StreamExt;

#[tokio::main]
async fn main() -> Result<(), async_nats::Error> {
    let client = async_nats::connect("127.0.0.1:4222").await?;
    let jetstream = jetstream::new(client);

    let stream = jetstream
        .get_or_create_stream(stream::Config {
            name: "ORDERS".to_string(),
            subjects: vec!["orders.>".to_string()],
            ..Default::default()
        })
        .await?;

    for index in 1..=3 {
        jetstream
            .publish("orders.created", format!("order {index}").into())
            .await?
            .await?;
    }

    let consumer = stream
        .get_or_create_consumer(
            "shipping",
            pull::Config {
                durable_name: Some("shipping".to_string()),
                ..Default::default()
            },
        )
        .await?;

    let mut messages = consumer.messages().await?;
    for _ in 0..3 {
        let message = messages.next().await.unwrap()?;
        println!("{} -> {}", message.message.subject, String::from_utf8_lossy(&message.message.payload));
        message.ack().await?;
    }
    Ok(())
}

Output

orders.created -> order 1
orders.created -> order 2
orders.created -> order 3

Without ack, the server redelivers a message after the ack wait time. The durable name lets the consumer resume after a restart.
Producers can set a Nats-Msg-Id header. The server drops repeats of the same id within a configurable window, which makes producer retries safe.

Example 5: Choosing a broker

Aspect	Kafka	RabbitMQ	NATS (JetStream)
Storage model	Partitioned log with retention	Queues, deleted on ack	Core: none. JetStream: streams
Ordering	Per partition	Per queue, weaker with several consumers	Per stream
Routing	Topic plus key	Exchanges, bindings, patterns	Subject wildcards
Replay	Yes, by offset	Not for classic queues (the separate Streams feature stores a log)	Yes, JetStream
Deployment	Cluster of brokers	One server, clustered for availability	One server binary, clustered for availability

Criteria:

Replay or several independent readers of the same history: a log (Kafka, JetStream).
Complex routing, per-message acks, priorities, or work queues: RabbitMQ.
Request-reply between services with no storage: NATS core.
Long retention of high-volume streams: Kafka.
Fewest components to operate: NATS, a single server binary.

Complete example

A POST /orders endpoint publishes a persistent, confirmed order.created message. The fulfillment worker from Example 2 consumes it.

Text

storefront-brokers/
  Cargo.toml
  src/
    lib.rs
    bin/orders_api.rs
    bin/rabbit_worker.rs
    bin/nats_core.rs
    bin/nats_jetstream.rs

Cargo.toml

TOML

[package]
name = "storefront-brokers"
version = "0.1.0"
edition = "2021"

[dependencies]
axum = "0.8"
lapin = "2.5"
tokio-executor-trait = "2.1"
tokio-reactor-trait = "1.1"
async-nats = "0.38"
tokio = { version = "1", features = ["full"] }
futures-util = "0.3"
serde = { version = "1", features = ["derive"] }
serde_json = "1"

src/lib.rs

Rust

use lapin::{
    options::{
        ExchangeDeclareOptions, QueueBindOptions, QueueDeclareOptions,
    },
    types::{AMQPValue, FieldTable},
    Channel, Connection, ConnectionProperties, ExchangeKind,
};

pub const URI: &str = "amqp://guest:guest@127.0.0.1:5672/%2f";

fn durable_exchange() -> ExchangeDeclareOptions {
    ExchangeDeclareOptions {
        durable: true,
        ..Default::default()
    }
}

fn durable_queue() -> QueueDeclareOptions {
    QueueDeclareOptions {
        durable: true,
        ..Default::default()
    }
}

pub async fn connect() -> lapin::Result<Connection> {
    let properties = ConnectionProperties::default()
        .with_executor(tokio_executor_trait::Tokio::current())
        .with_reactor(tokio_reactor_trait::Tokio);
    Connection::connect(URI, properties).await
}

pub async fn declare(channel: &Channel) -> lapin::Result<()> {
    channel
        .exchange_declare("orders", ExchangeKind::Topic, durable_exchange(), FieldTable::default())
        .await?;
    channel
        .exchange_declare("orders.dlx", ExchangeKind::Fanout, durable_exchange(), FieldTable::default())
        .await?;

    channel
        .queue_declare("orders.dead", durable_queue(), FieldTable::default())
        .await?;
    channel
        .queue_bind("orders.dead", "orders.dlx", "", QueueBindOptions::default(), FieldTable::default())
        .await?;

    let mut arguments = FieldTable::default();
    arguments.insert(
        "x-dead-letter-exchange".into(),
        AMQPValue::LongString("orders.dlx".into()),
    );
    channel
        .queue_declare("fulfillment", durable_queue(), arguments)
        .await?;
    channel
        .queue_bind("fulfillment", "orders", "order.*", QueueBindOptions::default(), FieldTable::default())
        .await?;

    channel
        .queue_declare("audit", durable_queue(), FieldTable::default())
        .await?;
    channel
        .queue_bind("audit", "orders", "order.#", QueueBindOptions::default(), FieldTable::default())
        .await?;
    Ok(())
}

src/bin/rabbit_worker.rs

Rust

use futures_util::StreamExt;
use lapin::{
    options::{BasicAckOptions, BasicConsumeOptions, BasicNackOptions, BasicQosOptions},
    types::FieldTable,
};
use storefront_brokers::{connect, declare};

#[tokio::main]
async fn main() -> lapin::Result<()> {
    let connection = connect().await?;
    let channel = connection.create_channel().await?;
    declare(&channel).await?;
    channel.basic_qos(10, BasicQosOptions::default()).await?;

    let mut consumer = channel
        .basic_consume(
            "fulfillment",
            "worker-1",
            BasicConsumeOptions::default(),
            FieldTable::default(),
        )
        .await?;

    while let Some(delivery) = consumer.next().await {
        let delivery = delivery?;
        match serde_json::from_slice::<serde_json::Value>(&delivery.data) {
            Ok(value) => {
                println!("handled {} {}", delivery.routing_key, value["order_id"]);
                delivery.ack(BasicAckOptions::default()).await?;
            }
            Err(_) => {
                println!("rejected {}", delivery.routing_key);
                delivery
                    .nack(BasicNackOptions {
                        requeue: false,
                        ..Default::default()
                    })
                    .await?;
            }
        }
    }
    Ok(())
}

src/bin/orders_api.rs

Rust

use axum::{extract::State, http::StatusCode, routing::post, Json, Router};
use lapin::{
    options::{BasicPublishOptions, ConfirmSelectOptions},
    publisher_confirm::Confirmation,
    BasicProperties, Channel,
};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use storefront_brokers::{connect, declare};

#[derive(Deserialize, Serialize)]
struct NewOrder {
    order_id: u64,
}

#[derive(Clone)]
struct AppState {
    channel: Channel,
}

async fn publish(channel: &Channel, order: &NewOrder) -> lapin::Result<bool> {
    let body = serde_json::to_vec(order).unwrap();
    let confirm = channel
        .basic_publish(
            "orders",
            "order.created",
            BasicPublishOptions::default(),
            &body,
            BasicProperties::default().with_delivery_mode(2),
        )
        .await?
        .await?;
    Ok(matches!(confirm, Confirmation::Ack(_)))
}

async fn create_order(
    State(state): State<AppState>,
    Json(order): Json<NewOrder>,
) -> (StatusCode, Json<Value>) {
    match publish(&state.channel, &order).await {
        Ok(true) => (
            StatusCode::ACCEPTED,
            Json(json!({ "order_id": order.order_id, "queued": true })),
        ),
        _ => (
            StatusCode::SERVICE_UNAVAILABLE,
            Json(json!({ "order_id": order.order_id, "queued": false })),
        ),
    }
}

#[tokio::main]
async fn main() {
    let connection = connect().await.expect("rabbitmq connection");
    let channel = connection.create_channel().await.expect("channel");
    declare(&channel).await.expect("topology");
    channel
        .confirm_select(ConfirmSelectOptions::default())
        .await
        .expect("confirm mode");

    let app = Router::new()
        .route("/orders", post(create_order))
        .with_state(AppState { channel });
    let listener = tokio::net::TcpListener::bind("127.0.0.1:3000").await.unwrap();
    axum::serve(listener, app).await.unwrap();
    drop(connection);
}

The handler answers 202 only after the broker confirms the message, and 503 otherwise. The connection value stays alive until the server stops.

Run (start the two containers from Project setup first):

Terminal

cargo run --bin rabbit_worker     # terminal 1
cargo run --bin orders_api        # terminal 2

Request and response:

Terminal

curl -s -i -X POST http://127.0.0.1:3000/orders \
  -H 'content-type: application/json' -d '{"order_id":7}' | head -n 1
curl -s -X POST http://127.0.0.1:3000/orders \
  -H 'content-type: application/json' -d '{"order_id":8}'

Output

HTTP/1.1 202 Accepted
{"order_id":8,"queued":true}

The worker terminal prints:

Output

handled order.created 7
handled order.created 8

The two NATS programs need only the nats container.

src/bin/nats_core.rs

Rust

use futures_util::StreamExt;

#[tokio::main]
async fn main() -> Result<(), async_nats::Error> {
    let client = async_nats::connect("127.0.0.1:4222").await?;

    let mut orders = client.subscribe("orders.*").await?;
    let mut requests = client.subscribe("inventory.check").await?;

    let responder = client.clone();
    tokio::spawn(async move {
        while let Some(request) = requests.next().await {
            if let Some(reply) = request.reply {
                responder.publish(reply, "in_stock".into()).await.unwrap();
            }
        }
    });

    client.publish("orders.created", "order 7".into()).await?;
    client.publish("orders.paid", "order 7".into()).await?;
    for _ in 0..2 {
        let message = orders.next().await.unwrap();
        println!("{} -> {}", message.subject, String::from_utf8_lossy(&message.payload));
    }

    let reply = client.request("inventory.check", "sku-1".into()).await?;
    println!("reply: {}", String::from_utf8_lossy(&reply.payload));
    Ok(())
}

src/bin/nats_jetstream.rs

Rust

use async_nats::jetstream::{self, consumer::pull, stream};
use futures_util::StreamExt;

#[tokio::main]
async fn main() -> Result<(), async_nats::Error> {
    let client = async_nats::connect("127.0.0.1:4222").await?;
    let jetstream = jetstream::new(client);

    let stream = jetstream
        .get_or_create_stream(stream::Config {
            name: "ORDERS".to_string(),
            subjects: vec!["orders.>".to_string()],
            ..Default::default()
        })
        .await?;

    for index in 1..=3 {
        jetstream
            .publish("orders.created", format!("order {index}").into())
            .await?
            .await?;
    }

    let consumer = stream
        .get_or_create_consumer(
            "shipping",
            pull::Config {
                durable_name: Some("shipping".to_string()),
                ..Default::default()
            },
        )
        .await?;

    let mut messages = consumer.messages().await?;
    for _ in 0..3 {
        let message = messages.next().await.unwrap()?;
        println!("{} -> {}", message.message.subject, String::from_utf8_lossy(&message.message.payload));
        message.ack().await?;
    }
    Ok(())
}

Terminal

cargo run --bin nats_core
cargo run --bin nats_jetstream

Run nats_core before nats_jetstream on a fresh container: once the ORDERS stream exists, it also stores the orders.* messages that nats_core publishes, and the JetStream program would read those first. Their outputs are the ones shown in Examples 3 and 4.

Summary

RabbitMQ producers publish to exchanges. Bindings route copies into queues. Consumers read queues.
Topic exchanges match * for one word and # for zero or more.
A durable queue plus a persistent message survives a broker restart. Publisher confirms confirm that the broker took the message.
Acks delete messages. Unacked messages are redelivered. Prefetch bounds messages in flight.
nack with requeue loops forever on a permanent failure. A dead-letter exchange holds those messages.
Core NATS is at-most-once with no storage. JetStream adds streams, durable consumers, acks, and replay.
NATS queue groups balance work, and request-reply is built in.
Broker choice follows replay needs, routing needs, retention, and the number of components to operate.