# How to Implement Work Queues with Message Prefetch in Kourier

> Implement work queues with message prefetch in Kourier by calling basicQos count 1u on AMQPChannel before consuming. Ensure fair distribution across workers.

- Repository: [Guimauve Digital/kourier](https://github.com/guimauvedigital/kourier)
- Tags: how-to-guide
- Published: 2026-03-02

---

**To implement work queues with message prefetch in Kourier, call `basicQos(count = 1u)` on your `AMQPChannel` before consuming to restrict each worker to one unacknowledged message at a time, ensuring fair distribution across consumers.**

Kourier provides a coroutine-friendly AMQP client that wraps RabbitMQ protocol operations in idiomatic Kotlin. When building background task processors, you need durable queues, persistent messages, and controlled message prefetching to prevent fast workers from monopolizing the queue. This guide shows you how to implement work queues with message prefetch in Kourier using the actual source from the `guimauvedigital/kourier` repository.

## Understanding the Message Prefetch API

The prefetch mechanism in Kourier is exposed through the `basicQos` method, which maps directly to RabbitMQ's **Basic.Qos** frame. This setting controls how many messages the broker delivers to a consumer before waiting for acknowledgments.

### The `basicQos` Method Signature

In [`amqp-client/src/commonMain/kotlin/dev/kourier/amqp/channel/AMQPChannel.kt`](https://github.com/guimauvedigital/kourier/blob/main/amqp-client/src/commonMain/kotlin/dev/kourier/amqp/channel/AMQPChannel.kt), the interface declares:

```kotlin
suspend fun basicQos(
    count: UShort,
    global: Boolean = false,
): AMQPResponse.Channel.Basic.QosOk

```

The `count` parameter specifies the prefetch window—the maximum number of unacknowledged messages allowed per consumer (or per channel if `global = true`). According to the source code, the default `global = false` applies the limit per individual consumer, which is the standard pattern for work queues.

### How Kourier Builds the QoS Frame

The concrete implementation resides in [`amqp-client/src/commonMain/kotlin/dev/kourier/amqp/channel/DefaultAMQPChannel.kt`](https://github.com/guimauvedigital/kourier/blob/main/amqp-client/src/commonMain/kotlin/dev/kourier/amqp/channel/DefaultAMQPChannel.kt). When you invoke `basicQos`, the library constructs a protocol frame with `prefetchSize = 0` (disabling size-based limits) and your specified `prefetchCount`:

```kotlin
override suspend fun basicQos(
    count: UShort,
    global: Boolean,
): AMQPResponse.Channel.Basic.QosOk {
    val qos = Frame(
        channelId = id,
        payload = Frame.Method.Basic.Qos(
            prefetchSize = 0u,
            prefetchCount = count,
            global = global
        )
    )
    return writeAndWaitForResponse(qos)
}

```

Convenience overloads in [`Extensions.kt`](https://github.com/guimauvedigital/kourier/blob/main/Extensions.kt) allow you to call `channel.basicQos(count = 1u)` without specifying the global flag explicitly.

## Implementing the Work Queue Pattern

A work queue requires three coordinated components: durable infrastructure, persistent messages, and consumers that acknowledge only after successful processing.

### Producer: Durable Queues and Persistent Messages

The producer must declare a **durable queue** that survives broker restarts and send **persistent messages** (delivery mode 2) written to disk. In the Kourier tutorial at [`docs/tutorials/work-queues.md`](https://github.com/guimauvedigital/kourier/blob/main/docs/tutorials/work-queues.md), the `newTask` function demonstrates this setup:

```kotlin
import dev.kourier.amqp.connection.amqpConfig
import dev.kourier.amqp.connection.createAMQPConnection
import dev.kourier.amqp.properties
import kotlinx.coroutines.CoroutineScope

suspend fun enqueueTask(scope: CoroutineScope, payload: String) {
    val cfg = amqpConfig {
        server { host = "localhost" }
    }
    val conn = createAMQPConnection(scope, cfg)
    val ch = conn.openChannel()

    // Durable queue declaration
    ch.queueDeclare(
        name = "task_queue",
        durable = true,
        exclusive = false,
        autoDelete = false,
        arguments = emptyMap()
    )

    // Persistent message properties
    val props = properties { deliveryMode = 2u }

    ch.basicPublish(
        message = payload.toByteArray(),
        exchange = "",
        routingKey = "task_queue",
        properties = props
    )
    println("[x] Sent '$payload'")

    ch.close()
    conn.close()
}

```

Setting `deliveryMode = 2u` ensures RabbitMQ writes the message to disk, preventing loss during broker restarts.

### Consumer: Configuring Prefetch and Manual Acknowledgment

Each worker must limit its prefetch to one message and use manual acknowledgment. This pattern is implemented in the `worker` function from the tutorial and tested in [`WorkQueuesTest.kt`](https://github.com/guimauvedigital/kourier/blob/main/WorkQueuesTest.kt):

```kotlin
import dev.kourier.amqp.connection.amqpConfig
import dev.kourier.amqp.connection.createAMQPConnection
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.delay

suspend fun startWorker(
    scope: CoroutineScope,
    name: String,
    onMessage: suspend (String) -> Unit
) {
    val cfg = amqpConfig {
        server { host = "localhost" }
    }
    val conn = createAMQPConnection(scope, cfg)
    val ch = conn.openChannel()

    ch.queueDeclare(
        name = "task_queue",
        durable = true,
        exclusive = false,
        autoDelete = false,
        arguments = emptyMap()
    )

    // Limit prefetch to one message per consumer
    ch.basicQos(count = 1u, global = false)

    // Consume with manual acknowledgment (noAck = false)
    val consumer = ch.basicConsume(queue = "task_queue", noAck = false)

    for (delivery in consumer) {
        val msg = delivery.message.body.decodeToString()
        println("[$name] Received '$msg'")
        try {
            onMessage(msg)  // Process the task
        } finally {
            // Acknowledge only after successful processing
            ch.basicAck(delivery.message, multiple = false)
        }
    }
}

```

With `noAck = false`, the broker holds the message until `basicAck` is called. If the consumer crashes before acknowledging, RabbitMQ automatically re-queues the message for another worker.

## Why Prefetch Enables Fair Dispatch

Without prefetch limits, RabbitMQ dispatches messages round-robin regardless of worker speed. This leads to **uneven distribution**: a fast worker finishes its tasks quickly while a slow worker accumulates a backlog in its local buffer.

Setting `basicQos(count = 1)` changes the broker behavior:

- The broker delivers exactly one message to a consumer
- The broker **holds** subsequent messages for that consumer until it receives an acknowledgment
- A slow worker cannot receive new messages while processing its current task
- Work is distributed based on actual availability, not just connection order

This mechanism, as implemented in [`DefaultAMQPChannel.kt`](https://github.com/guimauvedigital/kourier/blob/main/DefaultAMQPChannel.kt), ensures that heavy tasks are distributed fairly across your worker pool rather than piling up on the first available consumer.

## Complete Integration Example

To see fair dispatch in action, run multiple workers with different processing speeds. The following pattern from [`WorkQueuesTest.kt`](https://github.com/guimauvedigital/kourier/blob/main/WorkQueuesTest.kt) demonstrates how prefetch prevents fast workers from stealing all messages:

```kotlin
import kotlinx.coroutines.*

fun main() = runBlocking {
    // Enqueue tasks with varying complexity
    launch {
        listOf("Task.", "Long task..", "Task.", "Task.", "Heavy task....")
            .forEach { enqueueTask(this, it) }
    }

    // Worker 1: Fast processor
    launch { 
        startWorker(this, "Fast-Worker") { delay(1000L) } 
    }

    // Worker 2: Slow processor (starts slightly later)
    launch { 
        delay(100)
        startWorker(this, "Slow-Worker") { delay(it.length * 1000L) } 
    }

    delay(30_000)  // Allow processing to complete
}

```

Because both workers have `basicQos(count = 1u)`, the fast worker cannot pull all messages at once. The broker waits for the slow worker to acknowledge its current heavy task before delivering the next message, resulting in balanced throughput.

## Summary

- **Use `basicQos(count = 1u)`** on every consumer channel to enforce fair dispatch and prevent worker overload.
- **Declare queues with `durable = true`** and publish with `deliveryMode = 2u` to ensure messages survive broker restarts.
- **Consume with `noAck = false`** and call `basicAck` only after successful processing to guarantee at-least-once delivery.
- The implementation resides in [`AMQPChannel.kt`](https://github.com/guimauvedigital/kourier/blob/main/AMQPChannel.kt) and [`DefaultAMQPChannel.kt`](https://github.com/guimauvedigital/kourier/blob/main/DefaultAMQPChannel.kt), with tutorials available in [`docs/tutorials/work-queues.md`](https://github.com/guimauvedigital/kourier/blob/main/docs/tutorials/work-queues.md).

## Frequently Asked Questions

### What is the difference between prefetch count and prefetch size in Kourier?

**Prefetch count** (the `count` parameter in `basicQos`) limits the number of unacknowledged messages per consumer. **Prefetch size** would limit the total byte size of those messages, but Kourier's implementation in [`DefaultAMQPChannel.kt`](https://github.com/guimauvedigital/kourier/blob/main/DefaultAMQPChannel.kt) hardcodes `prefetchSize = 0u`, disabling size-based limits. You control flow purely by message count using unsigned short values.

### Why should I use `basicQos(count = 1u)` instead of default round-robin dispatch?

**Default round-robin dispatch sends messages to consumers regardless of their current workload**, causing slow workers to accumulate backlogs in local buffers while fast workers sit idle. With `basicQos(count = 1u)`, the broker withholds messages from busy consumers until they acknowledge their current task, distributing work based on actual processing capacity rather than connection order.

### How does Kourier handle message re-delivery when a consumer crashes?

**If a consumer crashes before calling `basicAck`, RabbitMQ automatically re-queues the message.** Because Kourier consumers use `noAck = false` (manual acknowledgment), unacknowledged messages return to the queue immediately when the channel closes unexpectedly. Another worker then receives the message, ensuring at-least-once delivery semantics for your work queue.

### Can I set a global prefetch limit across all consumers on a channel?

**Yes, by passing `global = true` to `basicQos`.** However, the Kourier API defaults to `global = false`, which applies the limit per consumer. Global limits are rarely used for work queues because they create coordination bottlenecks across multiple workers; per-consumer limits provide better isolation and predictable memory usage for individual processors.