InMemory — Consumer Groups

Use this when you want to see how consumer groups scale workers to match queue depth. One hundred messages are published and processed by three to six workers in parallel, each simulating 5 ms of I/O — a practical demonstration of the min/max worker range and prefetch configuration, all in-process.

Prerequisites

No external services required. The in-memory backend runs entirely in-process.

Cargo feature: inmemory

Run

cargo run --example inmemory_consumer_groups --features inmemory

Expected output

Lines arrive in non-deterministic order because workers process messages concurrently. After all 100 messages are consumed you will see a final summary:

processed work id=0
processed work id=3
processed work id=1
...
processed work id=99
processed 100 messages

Source

//! Consumer group with N workers sharing a single queue's load.
 
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
 
use serde::{Deserialize, Serialize};
use tokio_util::sync::CancellationToken;
 
use shove::inmemory::{InMemoryConfig, InMemoryConsumerGroupConfig};
use shove::{
    Broker, ConsumerGroupConfig, InMemory, JsonCodec, MessageHandler, MessageMetadata, Outcome,
    Topic, TopologyBuilder,
};
 
#[derive(Debug, Clone, Serialize, Deserialize)]
struct Work {
    id: u32,
}
 
struct WorkTopic;
impl Topic for WorkTopic {
    type Message = Work;
    type Codec = JsonCodec;
    fn topology() -> &'static shove::QueueTopology {
        static T: std::sync::OnceLock<shove::QueueTopology> = std::sync::OnceLock::new();
        T.get_or_init(|| TopologyBuilder::new("work").dlq().build())
    }
}
 
#[derive(Clone)]
struct Worker {
    count: Arc<AtomicUsize>,
}
impl MessageHandler<WorkTopic> for Worker {
    type Context = ();
    async fn handle(&self, msg: Work, _: MessageMetadata, _: &()) -> Outcome {
        // Simulate some I/O.
        tokio::time::sleep(Duration::from_millis(5)).await;
        self.count.fetch_add(1, Ordering::Relaxed);
        println!("processed work id={}", msg.id);
        Outcome::Ack
    }
}
 
#[tokio::main]
async fn main() {
    tracing_subscriber::fmt::init();
 
    let broker = Broker::<InMemory>::new(InMemoryConfig::default())
        .await
        .expect("connect InMemory");
    broker
        .topology()
        .declare::<WorkTopic>()
        .await
        .expect("declare");
 
    let count = Arc::new(AtomicUsize::new(0));
 
    let mut group = broker.consumer_group();
    let c = count.clone();
    group
        .register::<WorkTopic, _>(
            ConsumerGroupConfig::new(
                InMemoryConsumerGroupConfig::new(3..=6).with_prefetch_count(4),
            ),
            move || Worker { count: c.clone() },
        )
        .await
        .expect("register");
 
    let publisher = broker.publisher().await.expect("publisher");
    for i in 0..100u32 {
        publisher
            .publish::<WorkTopic>(&Work { id: i })
            .await
            .expect("publish");
    }
 
    // Stop when all 100 messages have been processed, or after a deadline.
    let stop = CancellationToken::new();
    let waiter_stop = stop.clone();
    let waiter_count = count.clone();
    tokio::spawn(async move {
        let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
        while waiter_count.load(Ordering::Relaxed) < 100 && tokio::time::Instant::now() < deadline {
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        waiter_stop.cancel();
    });
 
    let signal_stop = stop.clone();
    let outcome = group
        .run_until_timeout(
            async move { signal_stop.cancelled().await },
            Duration::from_secs(5),
        )
        .await;
 
    println!("processed {} messages", count.load(Ordering::Relaxed));
    std::process::exit(outcome.exit_code());
}

Walkthrough

Topic and worker definition

WorkTopic has a minimal topology: a "work" queue with a DLQ and no sequencing. Worker simulates realistic I/O with tokio::time::sleep(Duration::from_millis(5)) before returning Outcome::Ack. The Arc<AtomicUsize> counter is shared across all worker instances so the main thread can observe the running total.

Dynamic worker range

InMemoryConsumerGroupConfig::new(3..=6) declares a minimum of 3 and a maximum of 6 concurrent workers. The consumer group starts workers up to the minimum immediately and may spawn additional workers (up to the maximum) as queue depth warrants. With 100 messages and 5 ms per handler, the group has meaningful work to justify scaling toward 6 workers. with_prefetch_count(4) lets each worker hold up to 4 messages at a time, reducing per-message scheduling overhead.

Shutdown pattern

A background task polls the count AtomicUsize every 20 ms and cancels the CancellationToken once all 100 messages have been processed — or after a 30-second hard deadline. The final println! reports the actual count consumed, which should always be 100 on a successful run.

Factory closure and `Clone`

group.register takes a factory move || Worker { count: c.clone() }. The closure is called once per spawned worker, so each worker gets its own Worker instance with a cloned Arc pointing to the shared counter. The Clone derive on Worker is what enables this — each call to the factory produces a fresh value, but all share the same underlying atomic.

What to try next

Narrow the range to new(1..=1) — all 100 messages are processed sequentially and total wall-clock time grows to roughly 500 ms.
Widen the range to new(1..=32) — measure how throughput scales as more workers saturate the queue.
Increase with_prefetch_count(4) to with_prefetch_count(16) — observe fewer scheduling round-trips per message at the cost of head-of-line blocking if a slow message stalls a worker.
See Guides: Groups for how the group adapts its worker count at runtime.