Work Queue
Distribute tasks across competing workers using queues
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
- KubeMQ server running on
localhost:50000 - Kotlin SDK installed (
implementation("io.kubemq.sdk:kubemq-sdk-kotlin:1.0.1"))
Code
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
- Tasks are sent to a queue channel; each task is delivered to exactly one worker.
- Workers pull batches of tasks using
receiveQueuesMessageswithmaxItemsfor 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
Was this page helpful?