RabbitMQ — Concurrent Pubsub

Use this when deciding whether to enable concurrent processing for I/O-bound handlers. A 100 ms handler processes 20 messages sequentially in ~2 s; the same handler with with_concurrent_processing processes them in ~0.2 s — a 10x speedup visible in the printed summary. The example also confirms that acknowledgements remain in-order regardless of which mode is active.

Prerequisites

Docker (a RabbitMQ testcontainer is spun up automatically)
Cargo feature: rabbitmq

Run

cargo run --example rabbitmq_concurrent_pubsub --features rabbitmq

Expected output

--- Sequential (prefetch_count=1) ---
  20 messages, 100ms handler delay, prefetch=1
  processed task 0
  processed task 1
  ...
  done: 2.1s (10 msg/s)

--- Concurrent (prefetch_count=20) ---
  20 messages, 100ms handler delay, prefetch=20
  processed task 0
  processed task 3
  ...
  done: 0.2s (107 msg/s)

--- Summary ---
  sequential:  2.1s (10 msg/s)
  concurrent:  0.2s (107 msg/s)
  speedup:     10.1x

  Messages are always acked in delivery order.
  Concurrent mode overlaps handler I/O within a single consumer.

Exact numbers vary by system; expect a speedup close to the prefetch count (20x theoretical).

Source

//! Concurrent consumption example.
//!
//! Demonstrates `ConsumerOptions::with_concurrent_processing` for non-blocking
//! message processing with in-order acknowledgement. Compares sequential vs
//! concurrent throughput with a slow handler, both driven through the generic
//! `Broker<RabbitMq>::consumer_supervisor()` path.
//!
//! Spins up a RabbitMQ testcontainer automatically (requires a running
//! Docker daemon):
//!
//!     cargo run --example rabbitmq_concurrent_pubsub --features rabbitmq
 
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
use std::time::{Duration, Instant};
 
use serde::{Deserialize, Serialize};
use shove::rabbitmq::RabbitMqConfig;
use shove::{
    Broker, ConsumerOptions, MessageHandler, MessageMetadata, Outcome, Publisher, RabbitMq, Topic,
    TopologyBuilder, define_topic,
};
use testcontainers::runners::AsyncRunner;
use testcontainers_modules::rabbitmq::RabbitMq as RabbitMqImage;
 
// ─── Message type ───────────────────────────────────────────────────────────
 
#[derive(Debug, Clone, Serialize, Deserialize)]
struct Task {
    id: u32,
    payload: String,
}
 
// ─── Topic ──────────────────────────────────────────────────────────────────
 
define_topic!(
    SlowTasks,
    Task,
    TopologyBuilder::new("ex-concurrent-tasks")
        .hold_queue(Duration::from_secs(5))
        .dlq()
        .build()
);
 
// ─── Handler ────────────────────────────────────────────────────────────────
 
#[derive(Clone)]
struct SlowTaskHandler {
    processed: Arc<AtomicU32>,
    signal: Arc<tokio::sync::Notify>,
    delay: Duration,
}
 
impl SlowTaskHandler {
    fn new(delay: Duration) -> Self {
        Self {
            processed: Arc::new(AtomicU32::new(0)),
            signal: Arc::new(tokio::sync::Notify::new()),
            delay,
        }
    }
 
    fn count(&self) -> u32 {
        self.processed.load(Ordering::Relaxed)
    }
 
    async fn wait_for(&self, target: u32, timeout: Duration) -> bool {
        let deadline = tokio::time::Instant::now() + timeout;
        loop {
            if self.count() >= target {
                return true;
            }
            tokio::select! {
                _ = self.signal.notified() => {}
                _ = tokio::time::sleep_until(deadline) => {
                    return self.count() >= target;
                }
            }
        }
    }
}
 
impl MessageHandler<SlowTasks> for SlowTaskHandler {
    type Context = ();
    async fn handle(&self, task: Task, _meta: MessageMetadata, _: &()) -> Outcome {
        // Simulate slow I/O (HTTP call, database query, etc.)
        tokio::time::sleep(self.delay).await;
        eprintln!("  processed task {}", task.id);
        self.processed.fetch_add(1, Ordering::Relaxed);
        self.signal.notify_waiters();
        Outcome::Ack
    }
}
 
// ─── Helpers ────────────────────────────────────────────────────────────────
 
async fn connect(amqp_port: u16) -> (Broker<RabbitMq>, Publisher<RabbitMq>) {
    let uri = format!("amqp://guest:guest@localhost:{amqp_port}/%2f");
    let broker = Broker::<RabbitMq>::new(RabbitMqConfig::new(&uri))
        .await
        .expect("failed to connect to RabbitMQ");
    broker
        .topology()
        .declare::<SlowTasks>()
        .await
        .expect("failed to declare topology");
    let publisher = broker
        .publisher()
        .await
        .expect("failed to create publisher");
    (broker, publisher)
}
 
async fn publish_tasks(publisher: &Publisher<RabbitMq>, count: u32) {
    let messages: Vec<Task> = (0..count)
        .map(|id| Task {
            id,
            payload: format!("task-{id}"),
        })
        .collect();
    publisher
        .publish_batch::<SlowTasks>(&messages)
        .await
        .unwrap();
}
 
async fn purge_queue(mgmt_port: u16) {
    // Purge via management API to start clean.
    let http = reqwest::Client::new();
    let _ = http
        .delete(format!(
            "http://localhost:{mgmt_port}/api/queues/%2F/{}/contents",
            SlowTasks::topology().queue()
        ))
        .basic_auth("guest", Some("guest"))
        .send()
        .await;
    // Also purge the hold queue.
    let _ = http
        .delete(format!(
            "http://localhost:{mgmt_port}/api/queues/%2F/{}-hold-5s/contents",
            SlowTasks::topology().queue()
        ))
        .basic_auth("guest", Some("guest"))
        .send()
        .await;
    // Brief pause for the purge to take effect.
    tokio::time::sleep(Duration::from_millis(200)).await;
}
 
async fn run_round(
    broker: &Broker<RabbitMq>,
    handler: SlowTaskHandler,
    prefetch: u16,
    concurrent: bool,
    msg_count: u32,
) -> Duration {
    let mut supervisor = broker.consumer_supervisor();
    supervisor
        .register::<SlowTasks, _>(
            handler.clone(),
            ConsumerOptions::<RabbitMq>::new()
                .with_prefetch_count(prefetch)
                .with_concurrent_processing(concurrent),
        )
        .expect("register");
 
    let start = Instant::now();
 
    // Run the supervisor until every expected message has been processed, then
    // drain (bounded).
    let signal_handler = handler.clone();
    let outcome = supervisor
        .run_until_timeout(
            async move {
                signal_handler
                    .wait_for(msg_count, Duration::from_secs(60))
                    .await;
            },
            Duration::from_secs(10),
        )
        .await;
    let duration = start.elapsed();
    assert!(
        outcome.is_clean(),
        "supervisor reported errors: {outcome:?}"
    );
    duration
}
 
// ─── Main ───────────────────────────────────────────────────────────────────
 
#[tokio::main]
async fn main() {
    let container = RabbitMqImage::default()
        .start()
        .await
        .expect("failed to start RabbitMQ container");
    let amqp_port = container
        .get_host_port_ipv4(5672)
        .await
        .expect("failed to read AMQP port");
    let mgmt_port = container
        .get_host_port_ipv4(15672)
        .await
        .expect("failed to read management port");
 
    let msg_count = 20u32;
    let handler_delay = Duration::from_millis(100);
    let prefetch = 20u16;
 
    let (broker, publisher) = connect(amqp_port).await;
 
    // ── Sequential run ──────────────────────────────────────────────────
 
    eprintln!("--- Sequential (prefetch_count=1) ---");
    eprintln!("  {msg_count} messages, {handler_delay:?} handler delay, prefetch=1");
 
    purge_queue(mgmt_port).await;
    publish_tasks(&publisher, msg_count).await;
 
    let handler = SlowTaskHandler::new(handler_delay);
    let sequential_dur = run_round(&broker, handler, 1, false, msg_count).await;
 
    let seq_throughput = msg_count as f64 / sequential_dur.as_secs_f64();
    eprintln!(
        "  done: {:.1}s ({:.0} msg/s)\n",
        sequential_dur.as_secs_f64(),
        seq_throughput
    );
 
    // ── Concurrent run ──────────────────────────────────────────────────
 
    eprintln!("--- Concurrent (prefetch_count={prefetch}) ---");
    eprintln!("  {msg_count} messages, {handler_delay:?} handler delay, prefetch={prefetch}");
 
    purge_queue(mgmt_port).await;
    publish_tasks(&publisher, msg_count).await;
 
    let handler = SlowTaskHandler::new(handler_delay);
    let concurrent_dur = run_round(&broker, handler, prefetch, true, msg_count).await;
 
    let conc_throughput = msg_count as f64 / concurrent_dur.as_secs_f64();
    eprintln!(
        "  done: {:.1}s ({:.0} msg/s)\n",
        concurrent_dur.as_secs_f64(),
        conc_throughput
    );
 
    // ── Summary ─────────────────────────────────────────────────────────
 
    let speedup = sequential_dur.as_secs_f64() / concurrent_dur.as_secs_f64();
    eprintln!("--- Summary ---");
    eprintln!(
        "  sequential:  {:.1}s ({:.0} msg/s)",
        sequential_dur.as_secs_f64(),
        seq_throughput
    );
    eprintln!(
        "  concurrent:  {:.1}s ({:.0} msg/s)",
        concurrent_dur.as_secs_f64(),
        conc_throughput
    );
    eprintln!("  speedup:     {speedup:.1}x");
    eprintln!();
    eprintln!("  Messages are always acked in delivery order.");
    eprintln!("  Concurrent mode overlaps handler I/O within a single consumer.");
 
    broker.close().await;
    drop(container);
}

Walkthrough

Handler design

SlowTaskHandler wraps an Arc<AtomicU32> counter and an Arc<tokio::sync::Notify> signal. The handler sleeps for self.delay (100 ms) to simulate an I/O operation, then increments the counter and notifies any waiters. The wait_for(target, timeout) method polls the counter and wakes on Notify, letting the main task stop the supervisor as soon as the last message is processed rather than waiting on a fixed timer.

Sequential vs concurrent modes

The example calls run_round twice. For the sequential round: prefetch=1 (at most one message in-flight) and concurrent=false. For the concurrent round: prefetch=20 and concurrent=true. The key option is:

ConsumerOptions::<RabbitMq>::new()
    .with_concurrent_processing(concurrent)

When true, the consumer spawns each handler invocation as a separate tokio task, allowing up to prefetch handlers to run simultaneously. Acknowledgements are still sent in delivery order, so the broker's message ordering guarantees are maintained.

Queue management between rounds

purge_queue calls the RabbitMQ Management HTTP API (DELETE /api/queues/%2F/{name}/contents) between the two rounds to ensure the second round starts from a clean slate. The hold queue (named {queue}-hold-5s) is also purged to avoid stale retries bleeding into the concurrent run.

Speedup calculation

After both rounds complete, the example computes speedup = sequential_duration / concurrent_duration. With 20 messages at 100 ms each and prefetch 20, theoretical peak is 20x; real numbers are slightly lower due to tokio scheduling overhead. assert!(outcome.is_clean()) verifies the supervisor drained cleanly (no error exits) after each round.

What to try next

Lower prefetch to 5 while keeping concurrent=true — the speedup should converge toward 5x since at most 5 handlers run at once.
Increase msg_count to 200 — the sequential time grows linearly while concurrent time stays roughly constant until the prefetch window fills.
Set handler_delay to 0 ms — the speedup collapses to ~1x because there is no I/O to overlap; sequential and concurrent modes perform identically for CPU-bound handlers.
See Concepts: Handlers for the in-order acknowledgement guarantee and how concurrent mode interacts with prefetch.