# Work Queue (/sdks/go/how-to/work-queue)



## Overview [#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 [#prerequisites]

* KubeMQ server running on `localhost:50000`
* Go SDK installed (`go get github.com/kubemq-io/kubemq-go/v2`)

## Code [#code]

```go title="main.go"
// 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 [#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 [#related]

* [Pattern overview](/learn/guides/choosing-a-pattern)
* [Go SDK Reference](/sdks/go/reference)
* [Fan-Out](/sdks/go/how-to/fan-out)
* [Request-Reply](/sdks/go/how-to/request-reply)