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.

receiveQueueMessages() with pollMaxMessages(5) pulls a batch bounded by that limit and blocks up to pollWaitTimeoutInSeconds 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. Each message must be settled with m.ack() — it stays invisible until acknowledged and comes back after the visibility timeout if the worker never confirms, which is what makes the pattern at-least-once rather than fire-and-forget.

Gotchas: a worker that pulls a full pollMaxMessages 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 pollWaitTimeoutInSeconds turns polling into a busy-loop that hammers the broker for empty results; too long delays workers noticing new work. And forgetting ack() after processing means the message is never actually removed — it just keeps coming back, even though the work already happened.

Prerequisites

• KubeMQ server running on localhost:50000
• Java SDK installed (implementation 'io.kubemq.sdk:kubemq-sdk-Java:3.1.1' (Gradle) or Maven dependency from Getting Started)

Code

package io.kubemq.example.patterns;

import io.kubemq.sdk.queues.*;
import java.util.UUID;

/**
 * Work Queue Pattern Example
 *
 * Demonstrates the competing-consumers (work queue) pattern using KubeMQ queues.
 */
public class WorkQueueExample {

    private static final String ADDRESS = "localhost:50000";
    private static final String CLIENT_ID = "java-patterns-work-queue-client";
    private static final String CHANNEL = "java-patterns.work-queue";

    public static void main(String[] args) throws InterruptedException {
        // Create a client connected to the KubeMQ server
        QueuesClient client = QueuesClient.builder().address(ADDRESS).clientId(CLIENT_ID).build();
        client.ping();
        // Create the queue channel
        client.createQueuesChannel(CHANNEL);

        // Send tasks to the work queue
        System.out.println("Sending 10 tasks to work queue...\n");
        for (int i = 1; i <= 10; i++) {
            client.sendQueueMessage(QueueMessage.builder()
                    .id(UUID.randomUUID().toString()).channel(CHANNEL)
                    .body(("Task #" + i).getBytes()).build());
        }

        // Worker 1 pulls first batch
        System.out.println("Worker 1 pulling batch...");
        QueuesPollResponse resp1 = client.receiveQueueMessages(QueuesPollRequest.builder()
                .channel(CHANNEL).pollMaxMessages(5).pollWaitTimeoutInSeconds(3).build());
        System.out.println("  Received: " + resp1.getMessages().size());
        resp1.getMessages().forEach(m -> { System.out.println("    " + new String(m.getBody())); m.ack(); });

        // Worker 2 pulls remaining batch
        System.out.println("\nWorker 2 pulling batch...");
        QueuesPollResponse resp2 = client.receiveQueueMessages(QueuesPollRequest.builder()
                .channel(CHANNEL).pollMaxMessages(5).pollWaitTimeoutInSeconds(3).build());
        System.out.println("  Received: " + resp2.getMessages().size());
        resp2.getMessages().forEach(m -> { System.out.println("    " + new String(m.getBody())); m.ack(); });

        System.out.println("\nAll tasks distributed and processed.");

        // Clean up resources
        client.deleteQueuesChannel(CHANNEL);
        client.close();
    }
}

How It Works

• sendQueueMessage() places each task on the durable java-patterns.work-queue channel; messages are stored by the broker until a consumer polls them.
• receiveQueueMessages() with pollMaxMessages(5) fetches up to 5 messages in one RPC; the SDK returns whatever is available up to that limit without waiting beyond pollWaitTimeoutInSeconds.
• Each QueueMessageReceived.ack() sends a per-message acknowledgement back to the broker so the message is removed from the queue; without ack() the message would be redelivered after the visibility timeout expires.
• Two sequential receiveQueueMessages() calls here simulate two independent workers; in production each worker would run in its own thread or process calling the same queue channel.

Related

• Pattern overview
• Java SDK Reference
• Fan-Out
• Request-Reply