A work queue distributes time-consuming tasks across many workers. Multiple consumers compete on one queue (competing consumers), and each task goes to exactly one worker. With manual acknowledgment and prefetch, work is dispatched fairly and survives worker crashes. This is the most direct expression of the connector's model: an AMQP queue named tasks maps straight onto the KubeMQ Queue channel amqp.default.tasks.

Overview

A producer publishes tasks to the default exchange with routing-key set to the queue name — the default exchange routes by queue name. Each worker calls basic.consume with manual ack and sets a prefetch (basic.qos) so the broker never dispatches more than N unacked messages to it at once. A worker acks only after the task is done; if it dies first, the unacked message is requeued and another worker picks it up.

Operation	AMQP action	KubeMQ mapping
Declare	`queue.declare("tasks", durable=true)`	KubeMQ Queue channel `amqp.default.tasks`
Publish	`basic.publish(exchange="", routing-key="tasks")`	`SendQueueMessage` (default exchange → queue name)
Consume	`basic.consume("tasks")` + `basic.qos(prefetch=N)`	Competing-consumer pull with prefetch window
Ack	`basic.ack(delivery-tag)`	Message removed from the queue
Requeue	unacked on disconnect, or `basic.reject(requeue=true)`	Redelivered to the queue tail (`Redelivered=true`)

How it works

Each task is moved to exactly one of the competing workers. The broker respects each worker's prefetch budget, so a slow worker is not flooded. An unacked task whose worker disconnects is redelivered to another worker.

Each task is dispatched to exactly one competing worker; basic.ack removes it from the queue, and an unacked task whose worker disconnects is redelivered to another worker.

Publish and consume

Each example declares a durable tasks queue, publishes a batch of tasks on a confirm channel (so no publish is lost to an early close — see the callout below), then drains the queue with a manual-ack consumer that sets prefetch=1. Every client reads the broker endpoint from KUBEMQ_AMQP_URL (default amqp://guest:guest@localhost:5672/).

package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"time"

	amqp "github.com/rabbitmq/amqp091-go"
)

const queueName = "tasks"
const total = 10

func amqpURL() string {
	if v := os.Getenv("KUBEMQ_AMQP_URL"); v != "" {
		return v
	}
	return "amqp://guest:guest@localhost:5672/"
}

func main() {
	conn, err := amqp.Dial(amqpURL())
	if err != nil {
		log.Fatalf("dial: %v", err)
	}
	defer func() { _ = conn.Close() }()

	ch, err := conn.Channel()
	if err != nil {
		log.Fatalf("channel: %v", err)
	}
	defer func() { _ = ch.Close() }()

	// Durable queue: queue "tasks" → KubeMQ channel amqp.default.tasks.
	q, err := ch.QueueDeclare(queueName, true, false, false, false, nil)
	if err != nil {
		log.Fatalf("declare queue: %v", err)
	}

	// Produce — publish on a confirm channel so every task is durably enqueued
	// before we move on (a plain publish + immediate close can be lost; gotcha #9).
	if err := ch.Confirm(false); err != nil {
		log.Fatalf("confirm select: %v", err)
	}
	confirms := ch.NotifyPublish(make(chan amqp.Confirmation, total))
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	for i := 0; i < total; i++ {
		if err := ch.PublishWithContext(ctx, "", q.Name, false, false, amqp.Publishing{
			ContentType:  "text/plain",
			DeliveryMode: amqp.Persistent,
			Body:         []byte(fmt.Sprintf("task-%02d", i)),
		}); err != nil {
			log.Fatalf("publish task %d: %v", i, err)
		}
	}
	for i := 0; i < total; i++ {
		if c := <-confirms; !c.Ack {
			log.Fatalf("task %d nacked by broker", c.DeliveryTag)
		}
	}
	log.Printf(" [x] Published %d tasks to %q", total, q.Name)

	// Consume — manual ack with prefetch=1 (fair dispatch); ack after the work.
	if err := ch.Qos(1, 0, false); err != nil {
		log.Fatalf("qos: %v", err)
	}
	deliveries, err := ch.Consume(q.Name, "worker", false, false, false, false, nil)
	if err != nil {
		log.Fatalf("consume: %v", err)
	}
	seen := 0
	for seen < total {
		select {
		case d := <-deliveries:
			fmt.Printf(" [worker] %s (redelivered=%v)\n", d.Body, d.Redelivered)
			if err := d.Ack(false); err != nil { // ack → removed from the queue
				log.Fatalf("ack: %v", err)
			}
			seen++
		case <-ctx.Done():
			log.Fatalf("timed out after %d/%d tasks", seen, total)
		}
	}
	log.Printf(" [✓] Drained all %d tasks", seen)

	if _, err := ch.QueueDelete(q.Name, false, false, false); err != nil {
		log.Printf("warning: queue delete: %v", err)
	}
}

import os

import pika

URL = os.environ.get("KUBEMQ_AMQP_URL", "amqp://guest:guest@localhost:5672/")
QUEUE = "tasks"
TOTAL = 10


def main() -> None:
    conn = pika.BlockingConnection(pika.URLParameters(URL))
    ch = conn.channel()
    # Durable queue: queue "tasks" → KubeMQ channel amqp.default.tasks.
    ch.queue_declare(queue=QUEUE, durable=True)

    # Produce — confirm mode makes each publish block until the broker acks it,
    # so all tasks are durably enqueued before we consume (gotcha #9).
    ch.confirm_delivery()
    for i in range(TOTAL):
        ch.basic_publish(
            exchange="",
            routing_key=QUEUE,
            body=f"task-{i:02d}".encode(),
            properties=pika.BasicProperties(content_type="text/plain", delivery_mode=2),
        )
    print(f" [x] Published {TOTAL} tasks to {QUEUE!r}")

    # Consume — manual ack with prefetch=1 (fair dispatch); ack after the work.
    ch.basic_qos(prefetch_count=1)
    seen = 0
    for method, _props, body in ch.consume(QUEUE, inactivity_timeout=30, auto_ack=False):
        if method is None:
            raise SystemExit(f"timed out after {seen}/{TOTAL} tasks")
        print(f" [worker] {body.decode()} (redelivered={method.redelivered})")
        ch.basic_ack(method.delivery_tag)  # ack → removed from the queue
        seen += 1
        if seen >= TOTAL:
            break
    ch.cancel()
    print(f" [x] Drained all {seen} tasks")

    ch.queue_delete(queue=QUEUE)
    ch.close()
    conn.close()


if __name__ == "__main__":
    main()

import java.nio.charset.StandardCharsets;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.MessageProperties;

public final class Main {
    private static final String QUEUE = "tasks";
    private static final int TOTAL = 10;

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        // The Java client parses a trailing "/" as an EMPTY vhost; normalize it
        // back to the default "/" vhost (→ KubeMQ vhost "default").
        factory.setUri(System.getenv().getOrDefault("KUBEMQ_AMQP_URL", "amqp://guest:guest@localhost:5672/"));
        if (factory.getVirtualHost() == null || factory.getVirtualHost().isEmpty()) {
            factory.setVirtualHost("/");
        }

        try (Connection connection = factory.newConnection();
                Channel channel = connection.createChannel()) {
            // Durable queue: queue "tasks" → KubeMQ channel amqp.default.tasks.
            channel.queueDeclare(QUEUE, true, false, false, null);

            // Produce — confirm mode so every task is durably enqueued before we
            // consume (a publish + immediate close can be lost; gotcha #9).
            channel.confirmSelect();
            for (int i = 0; i < TOTAL; i++) {
                String task = String.format("task-%02d", i);
                channel.basicPublish("", QUEUE, MessageProperties.PERSISTENT_TEXT_PLAIN,
                        task.getBytes(StandardCharsets.UTF_8));
            }
            channel.waitForConfirmsOrDie(30_000);
            System.out.println("[x] Published " + TOTAL + " tasks to '" + QUEUE + "'");

            // Consume — manual ack with prefetch=1 (fair dispatch).
            channel.basicQos(1);
            CountDownLatch done = new CountDownLatch(TOTAL);
            DeliverCallback onTask = (tag, delivery) -> {
                String body = new String(delivery.getBody(), StandardCharsets.UTF_8);
                System.out.println("[worker] " + body
                        + " (redelivered=" + delivery.getEnvelope().isRedeliver() + ")");
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); // removed
                done.countDown();
            };
            channel.basicConsume(QUEUE, false, onTask, tag -> { });

            if (!done.await(30, TimeUnit.SECONDS)) {
                throw new IllegalStateException("timed out draining the queue");
            }
            System.out.println("[x] Drained all " + TOTAL + " tasks");
            channel.queueDelete(QUEUE);
        }
    }
}

import amqp from "amqplib";

const QUEUE = "tasks";
const TOTAL = 10;

function url(): string {
  return process.env["KUBEMQ_AMQP_URL"] ?? "amqp://guest:guest@localhost:5672/";
}

async function main(): Promise<void> {
  const connection = await amqp.connect(url());

  // Produce — a confirm channel so every task is durably enqueued before we
  // consume (a plain publish + immediate close can be lost; gotcha #9).
  const prodCh = await connection.createConfirmChannel();
  // Durable queue: queue "tasks" → KubeMQ channel amqp.default.tasks.
  await prodCh.assertQueue(QUEUE, { durable: true });
  for (let i = 0; i < TOTAL; i++) {
    const body = `task-${String(i).padStart(2, "0")}`;
    prodCh.sendToQueue(QUEUE, Buffer.from(body), { persistent: true });
  }
  await prodCh.waitForConfirms();
  console.log(`[x] Published ${TOTAL} tasks to "${QUEUE}"`);
  await prodCh.close();

  // Consume — manual ack with prefetch=1 (fair dispatch).
  const ch = await connection.createChannel();
  await ch.assertQueue(QUEUE, { durable: true });
  await ch.prefetch(1);
  let seen = 0;
  await new Promise<void>((resolve) => {
    ch.consume(
      QUEUE,
      (msg) => {
        if (msg === null) return;
        console.log(`[worker] ${msg.content.toString()} (redelivered=${msg.fields.redelivered})`);
        ch.ack(msg); // ack → removed from the queue
        if (++seen === TOTAL) resolve();
      },
      { noAck: false },
    );
  });
  console.log(`[x] Drained all ${seen} tasks`);

  await ch.deleteQueue(QUEUE);
  await ch.close();
  await connection.close();
}

main().catch((err) => {
  console.error(err);
  process.exit(1);
});

using System.Text;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

const string queueName = "tasks";
const int total = 10;

static string Url() =>
    Environment.GetEnvironmentVariable("KUBEMQ_AMQP_URL") is { Length: > 0 } v
        ? v
        : "amqp://guest:guest@localhost:5672/";

var factory = new ConnectionFactory { Uri = new Uri(Url()) };
await using var connection = await factory.CreateConnectionAsync("work-queues");

// Produce — a confirm channel so every task is durably enqueued before we
// consume (a publish + immediate close can be lost; gotcha #9).
var confirmOpts = new CreateChannelOptions(
    publisherConfirmationsEnabled: true,
    publisherConfirmationTrackingEnabled: true);
await using var prodCh = await connection.CreateChannelAsync(confirmOpts);
// Durable queue: queue "tasks" → KubeMQ channel amqp.default.tasks.
await prodCh.QueueDeclareAsync(queueName, durable: true, exclusive: false, autoDelete: false);
for (var i = 0; i < total; i++)
{
    var props = new BasicProperties { ContentType = "text/plain", Persistent = true };
    await prodCh.BasicPublishAsync("", queueName, mandatory: false, basicProperties: props,
        body: Encoding.UTF8.GetBytes($"task-{i:D2}"));
}
Console.WriteLine($"[x] Published {total} tasks to '{queueName}'");

// Consume — manual ack with prefetch=1 (fair dispatch).
await using var ch = await connection.CreateChannelAsync();
await ch.BasicQosAsync(prefetchSize: 0, prefetchCount: 1, global: false);
var done = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var seen = 0;
var consumer = new AsyncEventingBasicConsumer(ch);
consumer.ReceivedAsync += async (_, ea) =>
{
    var body = Encoding.UTF8.GetString(ea.Body.Span);
    Console.WriteLine($"[worker] {body} (redelivered={ea.Redelivered})");
    await ch.BasicAckAsync(ea.DeliveryTag, multiple: false); // removed from the queue
    if (Interlocked.Increment(ref seen) == total) done.TrySetResult();
};
await ch.BasicConsumeAsync(queueName, autoAck: false, consumer: consumer);

using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
await done.Task.WaitAsync(cts.Token);
Console.WriteLine($"[x] Drained all {total} tasks");

await ch.QueueDeleteAsync(queueName, ifUnused: false, ifEmpty: false);

# frozen_string_literal: true
require "bunny"
require "amq/uri"

QUEUE = "tasks"
TOTAL = 10

opts = AMQ::URI.parse(ENV.fetch("KUBEMQ_AMQP_URL", "amqp://guest:guest@localhost:5672/"))
# A bare trailing "/" parses to the EMPTY vhost; coerce it to the default "/".
opts[:vhost] = "/" if opts[:vhost].nil? || opts[:vhost].to_s.empty?
conn = Bunny.new(opts)
conn.start

ch = conn.create_channel
# Durable queue: queue "tasks" → KubeMQ channel amqp.default.tasks.
queue = ch.queue(QUEUE, durable: true)

# Produce — confirm channel so the publishes can't be left buffered/dropped on
# an immediate close (gotcha #9).
ch.confirm_select
TOTAL.times { |i| ch.default_exchange.publish(format("task-%02d", i), routing_key: QUEUE, persistent: true) }
ch.wait_for_confirms
puts " [x] Published #{TOTAL} tasks to '#{QUEUE}'"

# Consume — manual ack with prefetch=1 (fair dispatch).
ch.prefetch(1)
seen = 0
done = Queue.new
queue.subscribe(manual_ack: true, block: false) do |di, _props, body|
  puts " [worker] #{body} (redelivered=#{di.redelivered})"
  ch.ack(di.delivery_tag) # ack → removed from the queue
  seen += 1
  done.push(:done) if seen >= TOTAL
end
done.pop
puts " [x] Drained all #{seen} tasks"

queue.delete
conn.close

use futures_lite::StreamExt;
use lapin::{
    options::{
        BasicAckOptions, BasicConsumeOptions, BasicPublishOptions, BasicQosOptions,
        ConfirmSelectOptions, QueueDeclareOptions, QueueDeleteOptions,
    },
    publisher_confirm::Confirmation,
    types::FieldTable,
    BasicProperties, Connection, ConnectionProperties,
};

const QUEUE: &str = "tasks";
const TOTAL: usize = 10;

fn amqp_url() -> String {
    let url = std::env::var("KUBEMQ_AMQP_URL")
        .unwrap_or_else(|_| "amqp://guest:guest@localhost:5672/".into());
    // lapin needs the default "/" vhost as "%2f"; the canonical URL ends in "/".
    match url.rsplit_once('@').map_or(url.as_str(), |(_, host)| host) {
        host if host.ends_with('/') && !host.to_lowercase().ends_with("/%2f") => format!("{url}%2f"),
        _ => url,
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let conn = Connection::connect(&amqp_url(), ConnectionProperties::default()).await?;
    let ch = conn.create_channel().await?;
    // Durable queue: queue "tasks" → KubeMQ channel amqp.default.tasks.
    ch.queue_declare(QUEUE, QueueDeclareOptions { durable: true, ..Default::default() }, FieldTable::default())
        .await?;

    // Produce — confirm mode + wait-for-ack so every task is durably enqueued
    // before we consume (a publish + immediate close can be lost; gotcha #9).
    ch.confirm_select(ConfirmSelectOptions::default()).await?;
    for i in 0..TOTAL {
        let confirm = ch
            .basic_publish(
                "",
                QUEUE,
                BasicPublishOptions::default(),
                format!("task-{i:02}").as_bytes(),
                BasicProperties::default().with_delivery_mode(2), // persistent
            )
            .await?
            .await?;
        assert!(!matches!(confirm, Confirmation::Nack(_)), "broker nacked task-{i:02}");
    }
    println!("[x] Published {TOTAL} tasks to '{QUEUE}'");

    // Consume — manual ack with prefetch=1 (fair dispatch).
    ch.basic_qos(1, BasicQosOptions::default()).await?;
    let mut deliveries = ch
        .basic_consume(QUEUE, "worker", BasicConsumeOptions::default(), FieldTable::default())
        .await?;
    let mut seen = 0usize;
    while seen < TOTAL {
        if let Some(delivery) = deliveries.next().await {
            let delivery = delivery?;
            println!(
                "[worker] {} (redelivered={})",
                String::from_utf8_lossy(&delivery.data),
                delivery.redelivered
            );
            delivery.ack(BasicAckOptions::default()).await?; // ack → removed
            seen += 1;
        }
    }
    println!("[x] Drained all {seen} tasks");

    ch.queue_delete(QUEUE, QueueDeleteOptions::default()).await?;
    conn.close(0, "done").await?;
    Ok(())
}

At-least-once delivery

Unacked deliveries are requeued when a worker disconnects, so no task is lost even if a worker dies ungracefully. The trade-off is that a task may be redelivered — a redelivered message arrives with Redelivered == true, and the same body can be processed more than once. Workers must therefore be idempotent. Exactly-once is not provided.

Requeue lands at the tail, not the head. A requeued message re-enters at the tail of the queue, not the head — a deliberate deviation from RabbitMQ classic head-requeue. Ordering after a redelivery therefore differs from classic RabbitMQ; do not rely on strict publish order once a redelivery has occurred.

Prefer basic.consume over basic.get, and confirm before closing. Polling with basic.get has a ~1-second latency floor on an empty queue — push-style basic.consume is the right tool for work queues. And a fire-and-forget basic.publish followed by an immediate channel/connection close can silently drop publishes still buffered in the connector (no client error). Every multi-message producer above uses a confirm channel and waits for all acks before closing.

Work Queues

Overview

How it works

Publish and consume

At-least-once delivery

Pub/Sub (fanout)

RPC

Channel mapping

On this page

Work Queues

Overview

How it works

Publish and consume

At-least-once delivery

Related

Pub/Sub (fanout)

RPC

Channel mapping

On this page