Work Queue

Overview

A work queue distributes a stream of tasks across a pool of workers so each task is handled exactly once, instead of every worker doing every task — the pattern you reach for whenever you need to parallelize processing (image resizing, batch jobs, background work) without coordinating which worker owns which item. The queue itself does that coordination: workers just keep polling, and the broker load-balances whatever is next in line across whichever workers happen to be asking.

receive_queue_messages(channel, max_messages, wait_timeout_seconds, auto_ack) pulls a batch bounded by max_messages and blocks up to wait_timeout_seconds if the queue is empty, so a worker long-polls instead of busy-looping or hanging forever. Delivery is competing-consumer: once one worker's call returns a message, no other worker gets it. The trailing auto_ack flag determines the delivery guarantee — false holds each message invisible until explicitly acknowledged, redelivering it after the visibility window if the worker crashes first (at-least-once); true marks it done the instant it's handed over (at-most-once).

Gotchas: a worker that pulls a full max_messages batch and then crashes before acking every item in it leaves the unacked ones to be redelivered — possibly to a different worker — so size batches to what you can safely redo. A short wait_timeout_seconds turns polling into a busy-loop that hammers the broker for empty results; too long delays workers noticing new work. And auto_ack = true trades safety for simplicity — fine for idempotent, low-value tasks, wrong for anything that must survive a worker crash mid-task.

Prerequisites

• KubeMQ server running on localhost:50000
• Rust SDK installed (cargo add kubemq)

Code

use kubemq::prelude::*;
use kubemq::QueueMessageBuilder;

#[tokio::main]
async fn main() -> kubemq::Result<()> {
    let client = KubemqClient::builder()
        .host("localhost")
        .port(50000)
        .build()
        .await?;

    let channel = "rust-patterns.work-queue";

    for i in 0..10 {
        let msg = QueueMessageBuilder::new()
            .channel(channel)
            .body(format!("task-{}", i).into_bytes())
            .build();
        client.send_queue_message(msg).await?;
    }
    println!("Enqueued 10 tasks");

    let worker1 = client.receive_queue_messages(channel, 5, 3, false).await?;
    println!("Worker-1 processed {} tasks", worker1.len());

    let worker2 = client.receive_queue_messages(channel, 5, 3, false).await?;
    println!("Worker-2 processed {} tasks", worker2.len());

    client.close().await?;
    Ok(())
}

How It Works

• Queue messages are point-to-point — each message is delivered to exactly one consumer.
• Multiple workers pull from the same queue for natural load balancing.
• Combine with dead-letter queues and expiration for production resilience.
• Review timeouts, channel names, and client IDs before running against shared environments.
• Run the program while the server from the prerequisites is available.

Related

• Queues overview
• Rust SDK Reference
• Configuration reference
• Fan-Out