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 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 call returns a message, no other worker gets it. Each message must be settled with msg.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 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 forgetting msg.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
• Node.js SDK installed (npm install kubemq-js)

Code

/**
 * Example: Work Queue (Competing Consumers) Pattern
 *
 * Demonstrates the competing consumers pattern where multiple workers
 * pull from the same queue, but each message is delivered to exactly one
 * worker. This enables horizontal scaling of message processing.
 *
 * Prerequisites:
 *   - KubeMQ server running on localhost:50000
 *
 * Run: npx tsx examples/patterns/work-queue.ts
 */
import { KubeMQClient, createQueueMessage } from 'kubemq-js';

async function main(): Promise<void> {
  const client = await KubeMQClient.create({
    address: 'localhost:50000',
    clientId: 'js-patterns-work-queue-client',
  });

  try {
    const channel = 'js-patterns.work-queue';

    // Enqueue several tasks.
    const tasks = [
      'resize-image-001.jpg',
      'resize-image-002.jpg',
      'resize-image-003.jpg',
      'resize-image-004.jpg',
      'resize-image-005.jpg',
      'resize-image-006.jpg',
    ];

    for (const task of tasks) {
      await client.sendQueueMessage(
        createQueueMessage({
          channel,
          body: task,
          tags: { type: 'image-resize' },
        }),
      );
    }
    console.log(`Enqueued ${tasks.length} tasks\n`);

    // Simulate two competing workers pulling from the same queue.
    // Each worker receives different messages — no duplication.
    async function runWorker(name: string, count: number): Promise<void> {
      const messages = await client.receiveQueueMessages({
        channel,
        maxMessages: count,
        waitTimeoutSeconds: 5,
      });

      for (const msg of messages) {
        const body = new TextDecoder().decode(msg.body);
        console.log(`[${name}] Processing: ${body}`);
        // Simulate processing time.
        await new Promise((r) => setTimeout(r, 100));
        await msg.ack();
        console.log(`[${name}] Completed: ${body}`);
      }

      console.log(`[${name}] Finished — processed ${messages.length} tasks`);
    }

    // Both workers pull concurrently — each message goes to exactly one worker.
    await Promise.all([runWorker('Worker-A', 3), runWorker('Worker-B', 3)]);

    console.log('\nAll tasks processed by competing consumers');
  } finally {
    await client.close();
  }
}

main().catch(console.error);

How It Works

• 6 tasks are enqueued with sendQueueMessage(), then Promise.all([runWorker('Worker-A', 3), runWorker('Worker-B', 3)]) runs two workers concurrently, each pulling up to 3 messages — demonstrating competing-consumer semantics where each message is delivered to exactly one worker.
• receiveQueueMessages({ maxMessages: 3, waitTimeoutSeconds: 5 }) is a poll-and-hold: it blocks up to 5 seconds waiting for messages and returns a batch of up to 3.
• Each message is explicitly acknowledged with await msg.ack() after processing — without ack the message remains visible and gets redelivered after the server's visibility timeout.
• The two workers run concurrently via Promise.all; the total of 6 messages is split between them (not duplicated), which is the queue guarantee.

Related

• Pattern overview
• Node.js SDK Reference
• Fan-Out
• Request-Reply