RabbitMQ #

RabbitMQ is a message broker implementing the AMQP (Advanced Message Queuing Protocol). Unlike Kafka which stores events in an append-only log, RabbitMQ is a classic message broker: messages are sent to an exchange, routed to a queue, consumed by a consumer, and deleted after being acknowledged. This makes it the ideal choice for task queues, asynchronous RPC, and complex message routing. In Rust, the lapin crate is an async, actively developed AMQP 0-9-1 implementation. This article covers all RabbitMQ primitives — exchange, queue, binding, publish, consume — along with production patterns like Dead Letter Exchanges and RPC.

RabbitMQ Basic Concepts #

flowchart LR
    P["Producer"] --> EX

    subgraph RabbitMQ
        EX["Exchange\n(Direct/Fanout/Topic/Headers)"]
        EX -->|routing key| Q1["Queue A"]
        EX -->|routing key| Q2["Queue B"]
        EX -->|routing key| Q3["Queue C"]
    end

    Q1 --> C1["Consumer 1"]
    Q2 --> C2["Consumer 2"]
    Q3 --> C3["Consumer 3"]
ComponentFunction
ExchangeReceives messages from producers and forwards them to queues based on rules
QueueStores messages until consumed
BindingThe rule connecting an exchange to a queue
Routing KeyA label on a message matched against bindings
Ack/NackProcessing confirmation from the consumer

The Four Exchange Types #

ExchangeRoutingWhen to use
DirectExact routing key matchTask queues, specific routing
FanoutSend to all bound queuesBroadcasting, notifications
TopicWildcard pattern matching (* and #)Log routing, event categories
HeadersMessage header matchingComplex routing without routing keys

Installation #

[dependencies]
lapin = "2"
tokio = { version = "1", features = ["full"] }
tokio-amqp = "2"
serde = { version = "1", features = ["derive"] }
serde_json = "1"
futures = "0.3"

Connections and Channels #

use lapin::{
    options::*, types::FieldTable, Connection, ConnectionProperties,
};

async fn buat_koneksi(uri: &str) -> Result<Connection, lapin::Error> {
    Connection::connect(
        uri,
        ConnectionProperties::default()
            .with_executor(tokio_executor_trait::Tokio::current())
            .with_reactor(tokio_reactor_trait::Tokio),
    )
    .await
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Format: amqp://user:***@host:port/vhost
    let uri = std::env::var("RABBITMQ_URL")
        .unwrap_or_else(|_| "amqp://guest:***@localhost:5672/%2f".to_string());

    let conn = buat_koneksi(&uri).await?;
    println!("Connected to RabbitMQ");

    // Channel: the unit of work within one connection
    // One connection can have many channels (more efficient than many connections)
    let channel = conn.create_channel().await?;
    println!("Channel created");

    Ok(())
}

Declaring Exchanges and Queues #

Declarations are idempotent — safe to call many times as long as the parameters are the same:

use lapin::{
    options::*,
    types::{FieldTable, AMQPValue},
    Channel, ExchangeKind,
};

async fn setup_infrastruktur(channel: &Channel) -> Result<(), lapin::Error> {
    // Declare a Direct exchange
    channel.exchange_declare(
        "pesanan",                          // exchange name
        ExchangeKind::Direct,
        ExchangeDeclareOptions {
            durable: true,                  // survives broker restarts
            ..Default::default()
        },
        FieldTable::default(),
    )
    .await?;

    // Declare a Topic exchange for pattern-based routing
    channel.exchange_declare(
        "events",
        ExchangeKind::Topic,
        ExchangeDeclareOptions { durable: true, ..Default::default() },
        FieldTable::default(),
    )
    .await?;

    // Declare a Fanout exchange for broadcasting
    channel.exchange_declare(
        "notifikasi",
        ExchangeKind::Fanout,
        ExchangeDeclareOptions { durable: true, ..Default::default() },
        FieldTable::default(),
    )
    .await?;

    // Declare a Dead Letter Exchange (DLX) — for failed messages
    channel.exchange_declare(
        "dlx",
        ExchangeKind::Direct,
        ExchangeDeclareOptions { durable: true, ..Default::default() },
        FieldTable::default(),
    )
    .await?;

    // Declare a queue with a DLX
    let mut args = FieldTable::default();
    args.insert(
        "x-dead-letter-exchange".into(),
        AMQPValue::LongString("dlx".into()),
    );
    args.insert(
        "x-message-ttl".into(),
        AMQPValue::LongUInt(300_000), // 5 minute TTL
    );

    channel.queue_declare(
        "pesanan.baru",
        QueueDeclareOptions {
            durable: true,              // queue survives restarts
            ..Default::default()
        },
        args,
    )
    .await?;

    // DLQ queue to receive dead messages
    channel.queue_declare(
        "dlq.pesanan",
        QueueDeclareOptions { durable: true, ..Default::default() },
        FieldTable::default(),
    )
    .await?;

    // Bind the queue to the exchange with a routing key
    channel.queue_bind(
        "pesanan.baru",     // queue name
        "pesanan",          // exchange name
        "baru",             // routing key
        QueueBindOptions::default(),
        FieldTable::default(),
    )
    .await?;

    // Bind the DLQ to the DLX
    channel.queue_bind(
        "dlq.pesanan",
        "dlx",
        "pesanan.baru",
        QueueBindOptions::default(),
        FieldTable::default(),
    )
    .await?;

    println!("Exchanges and queues declared successfully");
    Ok(())
}

Publisher — Sending Messages #

use lapin::{
    options::BasicPublishOptions,
    BasicProperties, Channel,
};
use serde::{Deserialize, Serialize};

#[derive(Debug, Serialize, Deserialize, Clone)]
struct PesanPesanan {
    pub id: u64,
    pub pengguna_id: u64,
    pub produk: Vec<String>,
    pub total: f64,
    pub prioritas: u8,
}

struct Publisher {
    channel: Channel,
}

impl Publisher {
    fn baru(channel: Channel) -> Self {
        Publisher { channel }
    }

    async fn kirim<T: Serialize>(
        &self,
        exchange: &str,
        routing_key: &str,
        pesan: &T,
        prioritas: u8,
    ) -> Result<(), lapin::Error> {
        let payload = serde_json::to_vec(pesan)
            .map_err(|_| lapin::Error::InvalidChannel(0))?;

        let properties = BasicProperties::default()
            .with_content_type("application/json".into())
            .with_delivery_mode(2)      // 2 = persistent (survives restarts)
            .with_priority(prioritas)   // priority 0-255
            .with_message_id(uuid::Uuid::new_v4().to_string().into());

        self.channel
            .basic_publish(
                exchange,
                routing_key,
                BasicPublishOptions {
                    mandatory: true,    // error if no queue receives the message
                    ..Default::default()
                },
                &payload,
                properties,
            )
            .await?
            .await?;  // wait for broker confirmation (publisher confirm)

        Ok(())
    }
}

use uuid;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let conn = buat_koneksi("amqp://guest:***@localhost:5672/%2f").await?;
    let channel = conn.create_channel().await?;

    // Enable publisher confirm — wait for the broker to acknowledge each message
    channel.confirm_select(ConfirmSelectOptions::default()).await?;

    setup_infrastruktur(&channel).await?;

    let publisher = Publisher::baru(channel);

    let pesanan = PesanPesanan {
        id: 1001,
        pengguna_id: 42,
        produk: vec!["Laptop".to_string(), "Mouse".to_string()],
        total: 15_250_000.0,
        prioritas: 5,
    };

    // Send to the "pesanan" exchange with routing key "baru"
    publisher.kirim("pesanan", "baru", &pesanan, 5).await?;
    println!("Message sent: order #{}", pesanan.id);

    Ok(())
}

async fn buat_koneksi(uri: &str) -> Result<lapin::Connection, lapin::Error> {
    lapin::Connection::connect(
        uri,
        lapin::ConnectionProperties::default(),
    )
    .await
}

async fn setup_infrastruktur(_: &lapin::Channel) -> Result<(), lapin::Error> {
    Ok(()) // implementation above
}

Consumer — Receiving Messages #

Consumer with Manual Ack #

use futures::StreamExt;
use lapin::{
    message::DeliveryResult,
    options::*,
    Channel,
};

async fn konsumsi_pesanan(channel: Channel) -> Result<(), lapin::Error> {
    // QoS: max 10 messages in flight (prefetch)
    channel.basic_qos(10, BasicQosOptions::default()).await?;

    let mut consumer = channel
        .basic_consume(
            "pesanan.baru",             // queue name
            "consumer-pesanan-1",       // consumer tag (unique per consumer)
            BasicConsumeOptions {
                no_ack: false,          // manual ack (safer)
                ..Default::default()
            },
            FieldTable::default(),
        )
        .await?;

    println!("Consumer active, waiting for messages...");

    while let Some(delivery_result) = consumer.next().await {
        match delivery_result {
            Ok(delivery) => {
                let payload = std::str::from_utf8(&delivery.data)
                    .unwrap_or("");

                println!(
                    "Message received [tag: {}]: {}",
                    delivery.delivery_tag,
                    &payload[..payload.len().min(100)]
                );

                match serde_json::from_str::<PesanPesanan>(payload) {
                    Ok(pesanan) => {
                        match proses_pesanan(&pesanan).await {
                            Ok(_) => {
                                // Acknowledge — the message is removed from the queue
                                delivery
                                    .ack(BasicAckOptions::default())
                                    .await?;
                                println!("Order #{} processed", pesanan.id);
                            }
                            Err(e) => {
                                eprintln!("Failed to process order #{}: {}", pesanan.id, e);
                                // Nack with requeue=false → message goes to the DLX
                                delivery
                                    .nack(BasicNackOptions {
                                        requeue: false,
                                        ..Default::default()
                                    })
                                    .await?;
                            }
                        }
                    }
                    Err(e) => {
                        eprintln!("Message could not be parsed: {}", e);
                        // Corrupt message → discard (don't requeue)
                        delivery.reject(BasicRejectOptions { requeue: false }).await?;
                    }
                }
            }
            Err(e) => {
                eprintln!("Consumer error: {}", e);
                break;
            }
        }
    }

    Ok(())
}

async fn proses_pesanan(pesanan: &PesanPesanan) -> Result<(), String> {
    println!("Processing order #{} total Rp{:.0}", pesanan.id, pesanan.total);
    // Business logic...
    Ok(())
}

Parallel Consumer with Many Workers #

use std::sync::Arc;
use tokio::sync::Semaphore;

async fn konsumsi_paralel(
    channel: Channel,
    maks_paralel: usize,
) -> Result<(), lapin::Error> {
    // A semaphore limits parallel processing
    let semaphore = Arc::new(Semaphore::new(maks_paralel));

    channel.basic_qos(
        maks_paralel as u16,
        BasicQosOptions::default(),
    ).await?;

    let mut consumer = channel
        .basic_consume(
            "pesanan.baru",
            "consumer-paralel",
            BasicConsumeOptions { no_ack: false, ..Default::default() },
            FieldTable::default(),
        )
        .await?;

    while let Some(Ok(delivery)) = consumer.next().await {
        let permit = Arc::clone(&semaphore).acquire_owned().await.unwrap();

        tokio::spawn(async move {
            let _permit = permit; // the permit is released when the task finishes

            let payload = std::str::from_utf8(&delivery.data).unwrap_or("");
            if let Ok(pesanan) = serde_json::from_str::<PesanPesanan>(payload) {
                match proses_pesanan(&pesanan).await {
                    Ok(_) => { let _ = delivery.ack(BasicAckOptions::default()).await; }
                    Err(_) => {
                        let _ = delivery.nack(BasicNackOptions {
                            requeue: false, ..Default::default()
                        }).await;
                    }
                }
            } else {
                let _ = delivery.reject(BasicRejectOptions { requeue: false }).await;
            }
        });
    }

    Ok(())
}

use lapin::Channel;

async fn proses_pesanan(_: &PesanPesanan) -> Result<(), String> { Ok(()) }

#[derive(Debug, serde::Serialize, serde::Deserialize, Clone)]
struct PesanPesanan {
    id: u64,
    pengguna_id: u64,
    produk: Vec<String>,
    total: f64,
    prioritas: u8,
}

Topic Exchanges — Pattern-Based Routing #

Topic exchanges use wildcards for flexible routing:

async fn setup_topic_routing(channel: &Channel) -> Result<(), lapin::Error> {
    // The topic exchange was already declared earlier

    // Queue for all order events
    channel.queue_declare("log.pesanan", QueueDeclareOptions {
        durable: true, ..Default::default()
    }, FieldTable::default()).await?;

    // Queue for error events only
    channel.queue_declare("alert.error", QueueDeclareOptions {
        durable: true, ..Default::default()
    }, FieldTable::default()).await?;

    // Queue for all events from the payment service
    channel.queue_declare("log.pembayaran", QueueDeclareOptions {
        durable: true, ..Default::default()
    }, FieldTable::default()).await?;

    // Binding with topic patterns:
    // * = one word, # = zero or more words

    // pesanan.* → all order events (pesanan.baru, pesanan.selesai, etc.)
    channel.queue_bind("log.pesanan", "events", "pesanan.*",
        QueueBindOptions::default(), FieldTable::default()).await?;

    // *.error → all errors from all services
    channel.queue_bind("alert.error", "events", "*.error",
        QueueBindOptions::default(), FieldTable::default()).await?;

    // pembayaran.# → all events from the payment service
    channel.queue_bind("log.pembayaran", "events", "pembayaran.#",
        QueueBindOptions::default(), FieldTable::default()).await?;

    Ok(())
}

// Sending events with hierarchical routing keys
async fn kirim_event_topic(channel: &Channel) -> Result<(), lapin::Error> {
    let events = vec![
        ("pesanan.baru", r#"{"id": 1, "status": "baru"}"#),
        ("pesanan.selesai", r#"{"id": 1, "status": "selesai"}"#),
        ("pembayaran.berhasil", r#"{"id": 1, "jumlah": 150000}"#),
        ("pembayaran.gagal.timeout", r#"{"id": 2, "alasan": "timeout"}"#),
        ("inventori.error", r#"{"item": "A", "error": "stok habis"}"#),
    ];

    for (routing_key, payload) in events {
        channel.basic_publish(
            "events",
            routing_key,
            BasicPublishOptions::default(),
            payload.as_bytes(),
            BasicProperties::default().with_delivery_mode(2),
        )
        .await?
        .await?;

        println!("Event '{}' sent", routing_key);
    }

    Ok(())
}

RPC Pattern — Request-Reply via RabbitMQ #

RabbitMQ can be used for asynchronous RPC — a producer sends a request and waits for the reply in a temporary queue:

use lapin::{options::*, BasicProperties, Channel};
use futures::StreamExt;
use tokio::sync::oneshot;
use std::collections::HashMap;
use std::sync::{Arc, Mutex};

// RPC client
async fn rpc_call(
    channel: &Channel,
    payload: &str,
) -> Result<String, Box<dyn std::error::Error>> {
    // Create a temporary reply queue that will be deleted automatically
    let reply_queue = channel.queue_declare(
        "",  // empty name = broker generates a unique name
        QueueDeclareOptions {
            exclusive: true,    // only accessible by this connection
            auto_delete: true,  // deleted when the last consumer leaves
            ..Default::default()
        },
        FieldTable::default(),
    )
    .await?;

    let reply_queue_name = reply_queue.name().as_str().to_string();
    let correlation_id = uuid::Uuid::new_v4().to_string();

    // Send the request with correlation_id and reply_to
    channel.basic_publish(
        "",             // default exchange
        "rpc.hitung",  // routing to the RPC server queue
        BasicPublishOptions::default(),
        payload.as_bytes(),
        BasicProperties::default()
            .with_reply_to(reply_queue_name.clone().into())
            .with_correlation_id(correlation_id.clone().into()),
    )
    .await?
    .await?;

    // Wait for the reply in the temporary queue
    let mut consumer = channel.basic_consume(
        &reply_queue_name,
        "",
        BasicConsumeOptions {
            no_ack: true,
            ..Default::default()
        },
        FieldTable::default(),
    )
    .await?;

    // 5 second timeout
    let reply = tokio::time::timeout(
        std::time::Duration::from_secs(5),
        async {
            while let Some(Ok(delivery)) = consumer.next().await {
                let corr = delivery.properties.correlation_id()
                    .as_ref()
                    .map(|s| s.as_str().to_string())
                    .unwrap_or_default();

                if corr == correlation_id {
                    return Ok(String::from_utf8_lossy(&delivery.data).to_string());
                }
            }
            Err("No reply")
        },
    )
    .await
    .map_err(|_| "Timeout waiting for reply")??;

    Ok(reply)
}

RabbitMQ vs Kafka Comparison #

Aspect                  RabbitMQ                    Kafka

Storage model
  Delete after ack      Yes (default)                No — stored in a log
  Event replay          No (unless configured)       Yes — consumers re-read any offset

Routing
  Complexity            Very rich (4 exchange types) Simple (topic + partition)
  Wildcards             Yes (Topic exchange)         No

Throughput
  Per node              ~50k-100k msg/sec            ~1 million+ msg/sec
  Best for              Task queues, RPC             Event streaming, logs

Ordering
  Per queue             Yes (1 consumer)             Per partition

Acknowledgment
  Model                 Ack/Nack/Reject per message  Offset commit (batch)

Consumer
  Concurrency           Many consumers per queue     1 consumer per partition per group

Setup
  Complexity            Simple (single broker)      More complex (ZK/KRaft + broker)

Choose RabbitMQ when:
  ✓ Task queues — process jobs one by one with ack
  ✓ Asynchronous RPC — request-reply patterns
  ✓ Complex routing based on patterns or headers
  ✓ Dead Letter Queues that are easy to configure
  ✓ The team is already familiar with AMQP

Choose Kafka when:
  ✓ Very high-volume event streaming (millions/sec)
  ✓ Need historical event replay
  ✓ Event log as the source of truth
  ✓ Many independent consumer groups reading the same events

Summary #

  • A channel isn’t a connection — one AMQP connection can have many channels. Create one channel per thread/task, not one connection per thread.
  • Declarations are idempotent — queue_declare and exchange_declare are safe to call many times; the broker doesn’t error if it already exists with the same parameters.
  • delivery_mode: 2 for persistent messages — the message is flushed to disk before being acked. Without it, messages are lost if the broker restarts.
  • Manual ack is always safer than auto-ack — with no_ack: false, a message is only considered done after delivery.ack() is called. If the consumer crashes before the ack, the message will be requeued.
  • Nack with requeue: false → Dead Letter Exchange — if the queue is configured with a DLX, messages nacked without requeue are forwarded to the DLX for investigation.
  • Prefetch (basic_qos) for rate limiting — limit how many messages the broker sends to a consumer before an ack. Prevents consumers being overwhelmed with unbounded load.
  • Topic exchanges for flexible routing — pesanan.* matches one level, pembayaran.# matches zero or more levels. More powerful than Direct but still efficient.
  • Publisher confirms for delivery guarantees — enable confirm_select and await the basic_publish result twice to make sure the broker received the message.
  • RabbitMQ is easier, Kafka scales better — choose RabbitMQ for task queues and RPC, Kafka for high-volume event streaming that needs replay.

← Previous: Kafka   Next: Amazon SQS →

About | Author | Content Scope | Editorial Policy | Privacy Policy | Disclaimer | Contact