How to Implement Work Queues with Message Prefetch in Kourier

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, the interface declares:

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. When you invoke basicQos, the library constructs a protocol frame with prefetchSize = 0 (disabling size-based limits) and your specified prefetchCount:

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 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, the newTask function demonstrates this setup:

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:

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, 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 demonstrates how prefetch prevents fast workers from stealing all messages:

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 and DefaultAMQPChannel.kt, with tutorials available in 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 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}