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.

client.PollQueue pulls a batch bounded by MaxItems 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 — true tells the broker the message is done the instant it's handed over (at-most-once), while false holds it invisible until the worker calls dsMsg.Ack(), redelivering it after the visibility window if the worker never confirms (at-least-once).

Gotchas: a worker that pulls a full MaxItems batch and then crashes before acking loses — or, with manual ack, redelivers — every message in that batch, not just the one it was processing, 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
• Go SDK installed (go get github.com/kubemq-io/kubemq-go/v2)

Code

// Example: patterns/work-queue
//
// Demonstrates the work queue pattern using queues.
// Multiple messages are sent to a queue and consumed by workers.
// Each message is processed by exactly one worker (competing consumers).
//
// Channel: go-patterns.work-queue
// Client ID: go-patterns-work-queue-client
//
// Run with a KubeMQ server on localhost:50000
// (see https://docs.kubemq.io/deploy).
package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/kubemq-io/kubemq-go/v2"
)

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	client, err := kubemq.NewClient(ctx,
		kubemq.WithAddress("localhost", 50000),
		kubemq.WithClientId("go-patterns-work-queue-client"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	channel := "go-patterns.work-queue"

	// Producer: send multiple work items to the queue.
	for i := 1; i <= 5; i++ {
		msg := kubemq.NewQueueMessage().
			SetChannel(channel).
			SetBody(fmt.Appendf(nil, "task-%d", i)).
			SetMetadata(fmt.Sprintf("priority-%d", i))

		result, err := client.SendQueueMessage(ctx, msg)
		if err != nil {
			log.Fatal(err)
		}
		if result.IsError {
			log.Printf("Send error: %s", result.Error)
		} else {
			fmt.Printf("Enqueued: task-%d (id=%s)\n", i, result.MessageID)
		}
	}

	// Worker: consume and process work items via PollQueue.
	resp, err := client.PollQueue(ctx, &kubemq.PollRequest{
		Channel:            channel,
		MaxItems:           10,
		WaitTimeoutSeconds: 5,
		AutoAck:            true,
	})
	if err != nil {
		log.Fatal(err)
	}
	if resp.IsError {
		log.Fatalf("Receive failed: %s", resp.Error)
	}

	fmt.Printf("\nWorker processed %d tasks:\n", len(resp.Messages))
	for _, dsMsg := range resp.Messages {
		fmt.Printf("  - body=%s metadata=%s\n", dsMsg.Message.Body, dsMsg.Message.Metadata)
	}
}

How It Works

1. Five QueueMessage items are sent via client.SendQueueMessage; each returns a QueueSendResult with a broker-assigned MessageID.
2. client.PollQueue with MaxItems: 10 and WaitTimeoutSeconds: 5 retrieves up to 10 messages in a single round-trip, blocking for up to 5 seconds if the queue is empty.
3. AutoAck: true tells the broker to acknowledge all received messages immediately without requiring explicit Ack() calls — appropriate for at-most-once processing.
4. For at-least-once processing (retry on failure), use AutoAck: false and call dsMsg.Ack() per message after successful processing, or dsMsg.Nack() to return it to the queue.

Related

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