# Work Queue (/sdks/kotlin/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.

`receiveQueuesMessages` pulls a batch bounded by `maxItems` and blocks up to `waitTimeoutMs` 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. With `autoAck = false`, each message stays invisible until the worker calls `msg.ack()`, and comes back for redelivery 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 `maxItems` 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 `waitTimeoutMs` 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 [#prerequisites]

* KubeMQ server running on `localhost:50000`
* Kotlin SDK installed (`implementation("io.kubemq.sdk:kubemq-sdk-kotlin:1.0.1")`)

## Code [#code]

```kotlin title="WorkQueueExample.kt"
package io.kubemq.sdk.examples.patterns

import io.kubemq.sdk.client.KubeMQClient
import io.kubemq.sdk.queues.queueMessage
import kotlinx.coroutines.runBlocking

private const val ADDRESS = "localhost:50000"
private const val CLIENT_ID = "kotlin-patterns-work-queue"
private const val CHANNEL = "kotlin-patterns.work-queue"

fun main() = runBlocking {
    val client = KubeMQClient.queues {
        address = ADDRESS
        clientId = CLIENT_ID
    }

    client.use {
        try {
            client.createQueuesChannel(CHANNEL)

            // Send tasks to the work queue
            println("Sending 10 tasks to work queue...\n")
            repeat(10) { i ->
                client.sendQueuesMessage(queueMessage {
                    channel = CHANNEL
                    body = "Task #${i + 1}".toByteArray()
                })
            }

            // Worker 1 pulls first batch
            println("Worker 1 pulling batch...")
            val resp1 = client.receiveQueuesMessages {
                channel = CHANNEL
                maxItems = 5
                waitTimeoutMs = 3000
                autoAck = false
            }
            println("  Received: ${resp1.messages.size}")
            resp1.messages.forEach { msg ->
                println("    ${String(msg.body)}")
                msg.ack()
            }

            // Worker 2 pulls remaining batch
            println("\nWorker 2 pulling batch...")
            val resp2 = client.receiveQueuesMessages {
                channel = CHANNEL
                maxItems = 5
                waitTimeoutMs = 3000
                autoAck = false
            }
            println("  Received: ${resp2.messages.size}")
            resp2.messages.forEach { msg ->
                println("    ${String(msg.body)}")
                msg.ack()
            }

            println("\nAll tasks distributed and processed.")
        } finally {
            try { client.deleteQueuesChannel(CHANNEL) } catch (_: Exception) {}
        }
        println("Done.")
    }
}
```

## How It Works [#how-it-works]

* Tasks are sent to a queue channel; each task is delivered to exactly one worker.
* Workers pull batches of tasks using `receiveQueuesMessages` with `maxItems` for batch size control.
* Each worker acknowledges tasks after processing to prevent redelivery.
* If a worker fails to acknowledge, the task is redelivered to another worker.
* This pattern provides at-least-once delivery with load balancing across workers.

## Related [#related]

* [Fan-Out](/sdks/kotlin/how-to/fan-out)
* [Queues Reference](/sdks/kotlin/reference/queues)
* [Dead Letter Queue](/sdks/kotlin/how-to/queues/dead-letter-queue)