# Queues and Consumers (/connectors/rabbitmq/how-to/queues-and-consumers)



Every AMQP queue maps to a KubeMQ **Queue** channel `amqp.{vhost}.{queue}`. This guide covers
declaring queues, consuming, acknowledging (`ack` / `nack` / `reject`), prefetch (QoS), and
`basic.get`.

<Callout type="info">
  Consumption is **at-least-once**: unacked deliveries are requeued on disconnect, so there is **zero
  loss even on an ungraceful disconnect**. **Exactly-once is NOT provided** — plan for redelivery
  (`Redelivered == true`). See [Reliability](/connectors/rabbitmq/how-to/reliability).
</Callout>

## Declaring queues [#declaring-queues]

`queue.declare` supports `durable` / `exclusive` / `auto-delete` / `arguments`.

| Aspect                  | Behavior                                                                                                                                                                          |
| ----------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **Idempotency**         | Identical args → ok; mismatch → `406 precondition-failed`.                                                                                                                        |
| **Passive declare**     | `passive=true`: exists → ok; missing → `404 not-found`.                                                                                                                           |
| **Server-named queues** | `queue.declare("")` → the server mints `amq.gen-{uuid22}`.                                                                                                                        |
| **Exclusive queues**    | Connection-scoped. Cross-connection access (including passive declare) → `405 resource-locked`. Auto-deleted when the owning connection closes. &#x2A;*Node-local in a cluster.** |
| **Auto-delete queues**  | Deleted when the last (cluster-aware) consumer cancels.                                                                                                                           |

<Callout type="warn">
  **Exclusive queues are node-local.** In a cluster, an exclusive queue lives only on the node that
  owns the connection; cross-node access fails. For single-node deployments this is invisible. See
  [Reliability](/connectors/rabbitmq/how-to/reliability) and
  [Migration from RabbitMQ](/connectors/rabbitmq/reference/migration-from-rabbitmq).
</Callout>

## Consuming [#consuming]

`basic.consume` registers a consumer (an auto-generated `ctag-{n}` if the tag is empty). The
server then delivers `basic.deliver` + content header + body.

| Trigger                                    | Result                                                 |
| ------------------------------------------ | ------------------------------------------------------ |
| Duplicate consumer tag on a channel        | `530 not-allowed` (connection error, RabbitMQ dialect) |
| Exclusive consumer over existing consumers | `403 access-refused`                                   |

A minimal consume loop:

<Tabs groupId="language" items="['Go','Python','Java','JavaScript','C#','Ruby','Rust']">
  <Tab value="Go">
    ```go
    // amqp091-go — consume and manually ack each delivery.
    deliveries, _ := ch.Consume("orders", "", false /* autoAck */, false, false, false, nil)
    for d := range deliveries {
        process(d.Body)
        _ = d.Ack(false) // multiple=false
    }
    ```
  </Tab>

  <Tab value="Python">
    ```python
    # pika — consume with manual ack.
    def on_message(ch, method, props, body):
        process(body)
        ch.basic_ack(delivery_tag=method.delivery_tag)

    channel.basic_consume(queue="orders", on_message_callback=on_message, auto_ack=False)
    channel.start_consuming()
    ```
  </Tab>

  <Tab value="Java">
    ```java
    // amqp-client — consume with manual ack.
    DeliverCallback cb = (tag, delivery) -> {
        process(delivery.getBody());
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    };
    channel.basicConsume("orders", false /* autoAck */, cb, t -> {});
    ```
  </Tab>

  <Tab value="JavaScript">
    ```typescript
    // amqplib — consume with manual ack.
    await channel.consume("orders", (msg) => {
      if (!msg) return;
      process(msg.content);
      channel.ack(msg);
    }, { noAck: false });
    ```
  </Tab>

  <Tab value="C#">
    ```csharp
    // RabbitMQ.Client — consume with manual ack.
    var consumer = new EventingBasicConsumer(channel);
    consumer.Received += (_, ea) =>
    {
        Process(ea.Body.ToArray());
        channel.BasicAck(ea.DeliveryTag, multiple: false);
    };
    channel.BasicConsume("orders", autoAck: false, consumer);
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby
    # bunny — consume with manual ack.
    queue.subscribe(manual_ack: true, block: true) do |delivery_info, _props, body|
      process(body)
      channel.ack(delivery_info.delivery_tag)
    end
    ```
  </Tab>

  <Tab value="Rust">
    ```rust
    // lapin — consume with manual ack.
    let mut consumer = channel.basic_consume(
        "orders", "", BasicConsumeOptions::default(), FieldTable::default()).await?;
    while let Some(delivery) = consumer.next().await {
        let delivery = delivery?;
        process(&delivery.data);
        delivery.ack(BasicAckOptions::default()).await?;
    }
    ```
  </Tab>
</Tabs>

## Ack / nack / reject [#ack--nack--reject]

| Method                                            | Effect                                            |
| ------------------------------------------------- | ------------------------------------------------- |
| `basic.ack(tag, multiple)`                        | `AckRange` — message(s) consumed.                 |
| `basic.nack` / `basic.reject(tag, requeue=true)`  | `NAckRange` — requeued **at the tail**.           |
| `basic.reject` / `basic.nack(tag, requeue=false)` | Dropped, or dead-lettered if a DLX is configured. |
| Unknown delivery tag                              | `406 precondition-failed`.                        |

<Callout type="warn">
  **Requeue lands at the tail.** Requeued messages re-enter at the **queue tail**, not the head (a
  deviation from RabbitMQ classic head-requeue). Fairness ordering therefore differs.
</Callout>

### At-least-once consumption [#at-least-once-consumption]

Unacked deliveries are requeued on disconnect (a downstream nack-all safety net), so there is
**zero loss even on an ungraceful disconnect**. **Exactly-once is NOT provided** — plan for
redelivery (`Redelivered == true`).

## Prefetch (QoS) [#prefetch-qos]

`basic.qos(prefetch-size, prefetch-count, global)` limits in-flight unacked deliveries.

| Scope        | `global`                   | Meaning                                                   |
| ------------ | -------------------------- | --------------------------------------------------------- |
| Per-consumer | `false` (RabbitMQ default) | The budget applies to each consumer.                      |
| Per-channel  | `true`                     | The budget is shared across all consumers on the channel. |

* `prefetch-size` is accepted but **inert**.
* Default = **unlimited**.
* Effective allowance = `min(per-consumer budget, remaining channel-global budget)`.

## `basic.get` (pull) [#basicget-pull]

`basic.get` returns `get-ok` (with a delivery tag) or `get-empty`.

<Callout type="warn">
  **`basic.get` has a \~1s latency floor.** On an empty queue, `basic.get` blocks up to \~1 second (the
  KubeMQ minimum wait timeout) before returning `get-empty`. Polling with `basic.get` is therefore
  slow — &#x2A;*prefer `basic.consume`** for throughput. `GetBatchSize` (default 32) bounds per-`Get`
  pulls.
</Callout>

## Recover / flow [#recover--flow]

| Method                                                 | Behavior                                                          |
| ------------------------------------------------------ | ----------------------------------------------------------------- |
| `basic.recover(requeue=true)`                          | Nack-all unacked on the channel (tail requeue).                   |
| `basic.recover(requeue=false)` / `basic.recover-async` | `540 not-implemented`.                                            |
| `channel.flow`                                         | Replies `flow-ok`, takes no action (deprecated in the AMQP spec). |

## Purge / delete [#purge--delete]

* `queue.purge` returns the exact message count purged.
* `queue.delete` sends a server-initiated `basic.cancel(consumerTag)` to live consumers and
  returns the residual count.

## Related [#related]

<Cards>
  <Card title="Exchanges and routing" href="/connectors/rabbitmq/concepts/exchanges-and-routing" description="How a publish resolves through default/direct/fanout/topic/headers exchanges into these queues." />

  <Card title="Reliability" href="/connectors/rabbitmq/how-to/reliability" description="Publisher confirms, dead-letter exchanges, per-message TTL, and at-least-once delivery." />

  <Card title="Error codes" href="/connectors/rabbitmq/reference/error-codes" description="The 404, 405, 406, and 530 AMQP codes the connector returns on declare, consume, and ack failures." />
</Cards>