# Work Queues (/connectors/rabbitmq/how-to/work-queues)



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 [#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 [#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.

<Mermaid
  chart="`
graph LR
PROD[&#x22;Producer<br/>(publish key=tasks)&#x22;]
CONN[&#x22;RabbitMQ connector<br/>:5672&#x22;]
BROKER[&#x22;Message Broker&#x22;]
Q{{&#x22;Queue channel<br/>amqp.default.tasks&#x22;}}
W1[&#x22;Worker A<br/>(prefetch=1)&#x22;]
W2[&#x22;Worker B<br/>(prefetch=1)&#x22;]

PROD -- &#x22;basic.publish (default exchange)&#x22; --> CONN
CONN -- &#x22;SendQueueMessage&#x22; --> BROKER
BROKER --> Q
Q -- &#x22;one task<br/>(competing)&#x22; --> W1
Q -. &#x22;next task&#x22; .-> W2
W1 -. &#x22;basic.ack&#x22; .-> CONN

class PROD,W1,W2 client
class CONN connector
class BROKER broker
class Q queue
`"
/>

*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 [#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/`).

<Tabs groupId="language" items="['Go','Python','Java','JavaScript','C#','Ruby','Rust']">
  <Tab value="Go">
    ```go
    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)
    	}
    }
    ```
  </Tab>

  <Tab value="Python">
    ```python
    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()
    ```
  </Tab>

  <Tab value="Java">
    ```java
    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);
            }
        }
    }
    ```
  </Tab>

  <Tab value="JavaScript">
    ```typescript
    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);
    });
    ```
  </Tab>

  <Tab value="C#">
    ```csharp
    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);
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby
    # 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
    ```
  </Tab>

  <Tab value="Rust">
    ```rust
    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(())
    }
    ```
  </Tab>
</Tabs>

## At-least-once delivery [#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.

<Callout type="warn">
  **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.
</Callout>

<Callout type="info">
  **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.
</Callout>

## Related [#related]

<Cards>
  <Card title="Pub/Sub (fanout)" href="/connectors/rabbitmq/how-to/pub-sub" description="Broadcast every message to all subscribers via a fanout exchange." />

  <Card title="RPC" href="/connectors/rabbitmq/how-to/rpc" description="Native request/reply over amq.rabbitmq.reply-to — no gRPC responder." />

  <Card title="Channel mapping" href="/connectors/rabbitmq/reference/channel-mapping" description="The full amqp.{vhost}.{queue} grammar, naming rules, and the reserved default vhost." />
</Cards>