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.

PollAsync pulls a batch bounded by MaxMessages and blocks up to WaitTimeoutSeconds 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 poll call returns a message, no other worker gets it. AutoAck determines the delivery guarantee — false holds each message invisible in a transaction window until the worker calls msg.AckAsync(), redelivering it after the window expires if the worker crashes first (at-least-once); true would mark it done the instant it's handed over (at-most-once).

Gotchas: a worker that pulls a full MaxMessages 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 WaitTimeoutSeconds turns polling into a busy-loop that hammers the broker for empty results; too long delays workers noticing new work. And AutoAck = 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
• C# SDK installed (dotnet add package KubeMQ.SDK.CSharp)

Code

// KubeMQ .NET SDK — Patterns: Work Queue
//
// This example demonstrates the competing consumers (work queue) pattern using queues.
// Multiple workers poll the same queue, and each message is delivered to exactly one worker.
// This provides load balancing across workers.
// Uses QueueDownstreamReceiver.PollAsync for transactional message settlement.
//
// Prerequisites:
//   - KubeMQ server running on localhost:50000
//   - dotnet run

using KubeMQ.Sdk.Client;
using KubeMQ.Sdk.Queues;
using System.Text;

await using var client = new KubeMQClient(new KubeMQClientOptions
{
    ClientId = "csharp-patterns-work-queue-client",
});
await client.ConnectAsync();

Console.WriteLine("Connected to KubeMQ server");

// Enqueue work items
for (var i = 1; i <= 6; i++)
{
    await client.SendQueueMessageAsync(new QueueMessage
    {
        Channel = "csharp-patterns.work-queue",
        Body = Encoding.UTF8.GetBytes($"Task #{i}"),
    });
}

Console.WriteLine("Enqueued 6 tasks");

// Simulate 2 workers pulling from the queue, each with its own receiver
for (int worker = 1; worker <= 2; worker++)
{
    await using var receiver = await client.CreateQueueDownstreamReceiverAsync();

    var batch = await receiver.PollAsync(new QueuePollRequest
    {
        Channel = "csharp-patterns.work-queue",
        MaxMessages = 3,
        WaitTimeoutSeconds = 5,
        AutoAck = false,
    });

    Console.WriteLine($"\n[Worker-{worker}] Received {batch.Messages.Count} tasks:");
    foreach (var msg in batch.Messages)
    {
        Console.WriteLine($"  Processing: {Encoding.UTF8.GetString(msg.Body.Span)}");
        await msg.AckAsync();
    }
}

Console.WriteLine("\nDone.");

How It Works

• CreateQueueDownstreamReceiverAsync() returns a QueueDownstreamReceiver that opens a dedicated gRPC stream to the broker for transactional polling.
• PollAsync with AutoAck = false receives up to MaxMessages messages and holds them in a transaction window for WaitTimeoutSeconds; each message must be explicitly settled.
• await msg.AckAsync() removes the message from the queue permanently; if the worker crashes before acking, the broker redelivers after the visibility window expires.
• Each worker uses its own receiver instance — the broker distributes messages across competing consumers so no two workers receive the same message.

Related

• Pattern overview
• C# SDK Reference
• Fan-Out
• Request-Reply