Implementation:Fede1024 Rust rdkafka Async Worker Pool Pattern

Knowledge Sources	Tokio spawn rust-rdkafka
Domains	Concurrency, Stream_Processing
Last Updated	2026-02-07 19:00 GMT

Overview

Concrete concurrent worker pattern using tokio::spawn and FuturesUnordered as demonstrated in rust-rdkafka's async processing example.

Description

This is a Pattern Doc documenting the concurrent worker scaling pattern used in rust-rdkafka's examples/asynchronous_processing.rs. It combines tokio::spawn for launching worker tasks and FuturesUnordered for collecting and polling them.

The FutureProducer implements Clone via an internal Arc<ThreadedProducer>, making it cheap to share across workers. Each worker calls consumer.stream() or consumer.recv() to get messages, processes them (potentially via spawn_blocking), and produces results.

Usage

Use this pattern when you need to scale processing beyond a single async task. Clone the FutureProducer for each worker, spawn workers with tokio::spawn, and collect handles in a FuturesUnordered or Vec.

Code Reference

Source Location

• Repository: External pattern (tokio::spawn, futures::stream::FuturesUnordered)
• Example: examples/asynchronous_processing.rs in rust-rdkafka repository

Signature

// tokio::spawn
pub fn spawn<F>(future: F) -> JoinHandle<F::Output>
where
    F: Future + Send + 'static,
    F::Output: Send + 'static,

// FuturesUnordered
pub struct FuturesUnordered<Fut> { .. }
impl<Fut: Future> FuturesUnordered<Fut> {
    pub fn new() -> FuturesUnordered<Fut>;
    pub fn push(&self, future: Fut);
}

// FutureProducer Clone (cheap Arc clone)
impl<C, R> Clone for FutureProducer<C, R>
where
    C: ClientContext + 'static,
{
    fn clone(&self) -> FutureProducer<C, R>;
}

Import

use tokio;
use futures::stream::{FuturesUnordered, StreamExt};
use rdkafka::producer::FutureProducer;

I/O Contract

Inputs

Name	Type	Required	Description
num_workers	usize	Yes	Number of concurrent worker tasks
producer	FutureProducer	Yes	Shared producer (cloned per worker)
consumer	StreamConsumer	Yes	Shared consumer (stream() called per worker)

Outputs

Name	Type	Description
worker handles	Vec<JoinHandle<()>> or FuturesUnordered	Handles for all spawned workers

Usage Examples

Multi-Worker Pipeline

use futures::stream::{FuturesUnordered, StreamExt};
use rdkafka::consumer::{Consumer, StreamConsumer};
use rdkafka::producer::{FutureProducer, FutureRecord};
use rdkafka::Message;
use std::time::Duration;

let num_workers = 4;
let mut workers = FuturesUnordered::new();

for _ in 0..num_workers {
    let producer = producer.clone();
    let handle = tokio::spawn(async move {
        loop {
            let msg = consumer.recv().await.expect("recv failed");
            let owned = msg.detach();
            let processed = tokio::task::spawn_blocking(move || {
                expensive_transform(owned.payload().unwrap_or(&[]))
            })
            .await
            .expect("task panicked");

            producer
                .send(
                    FutureRecord::to("output-topic").payload(&processed),
                    Duration::from_secs(0),
                )
                .await;
        }
    });
    workers.push(handle);
}

// Wait for all workers
while let Some(result) = workers.next().await {
    result.expect("Worker panicked");
}

Related Pages

Implements Principle

• Principle:Fede1024_Rust_rdkafka_Concurrent_Worker_Scaling

Requires Environment

• Environment:Fede1024_Rust_rdkafka_Rust_Librdkafka_Build_Environment