Moving our RabbitMQ client from Actix to plain Tokio

I replaced an actor-heavy generated client with plain Tokio, bounded workers and batched confirms, then measured the result in production.

Updated 3 min read

I rewrote our generated RabbitMQ client from Actix actors to plain Tokio. The old version worked, and I have nothing against actors. The message path had simply collected more machinery than the problem needed, and it had become unpleasant to reason about during nightly incidents.

Too many layers

A delivery passed through supervisors, actor mailboxes, boxed handlers, futures tied to actor contexts and several layers of code before reaching a job that was basically: read the message, route it, run the handler and acknowledge it.

Connections and reconnects

The new version keeps those steps visible. Ready connections are read through ArcSwap, so the hot path doesn't need a shared write lock. Reconnects go through one gate: the first task that notices the broken connection does the work, and the others wait and recheck instead of all reconnecting at once.

Bounded consumers

Consumers run as a bounded worker pool. A semaphore limits in-flight messages to the prefetch size, a JoinSet owns active handlers, and shutdown takes priority over accepting more work. The clients are still typed and schema-generated; I only removed the actor layer between RabbitMQ and the handler.

A simplified version of the consumer loop (the real one is generated and typed per schema):

Rust
async fn consume<H: Handler>(
    mut deliveries: impl Stream<Item = Delivery> + Unpin,
    handler: Arc<H>,
    prefetch: usize,
    shutdown: CancellationToken,
) {
    let slots = Arc::new(Semaphore::new(prefetch));
    let mut active = JoinSet::new();

    loop {
        // Wait for a free slot before taking another message: at most `prefetch` in flight.
        let permit = tokio::select! {
            biased; // shutdown wins over new work
            _ = shutdown.cancelled() => break,
            permit = slots.clone().acquire_owned() => permit.expect("semaphore is never closed"),
        };
        let delivery = tokio::select! {
            biased;
            _ = shutdown.cancelled() => break,
            next = deliveries.next() => match next {
                Some(delivery) => delivery,
                None => break,
            },
        };

        let handler = handler.clone();
        active.spawn(async move {
            let _permit = permit; // frees the slot when the handler is done
            let outcome = handler.handle(&delivery).await;
            delivery.settle(outcome).await; // ack, or nack and requeue
        });
        while active.try_join_next().is_some() {}
    }

    // No new work; let the handlers already running finish and acknowledge.
    while active.join_next().await.is_some() {}
}

Backpressure falls out of the structure: when every slot is busy, the loop simply stops pulling from the broker.

Batched publisher confirms

The producer improved most after batching publisher confirms. A single publish still waits for its confirm, but a batch publishes several messages and waits once, so the round-trip is paid per batch instead of per message.

Production results

After rollout, consumer throughput went up by about 20% and producer throughput by roughly two to three times in messaging-heavy workloads. Memory dropped by 30–40% under heavy traffic and by a few megabytes per idle pod. Removing the Actix dependency tree cut service build time by around 10%.

Context for the numbers

The database is still the main bottleneck. "Producer 3x faster" doesn't mean the whole pricing platform got three times faster. It means one important layer adds less latency and causes fewer problems.

Fewer alerts

The result I care about most is less glamorous than the benchmark: recurring messaging-related alerts dropped to effectively zero. The generated client is now easier to inspect, bounded under load and no longer the thing waking me up at night.

Loading…