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.

Declaring queues

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

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.

A minimal consume loop:

// 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
}

# 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()

// 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 -> {});

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

// 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);

# 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

// 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?;
}

Ack / nack / reject

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)

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

• prefetch-size != 0 is refused 540 not-implemented — RabbitMQ refuses a byte-based prefetch rather than ignoring it.
• Default = unlimited.
• Effective allowance = min(per-consumer budget, remaining channel-global budget).

basic.get (pull)

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

Recover / flow

Purge / delete

• queue.purge returns the exact message count purged.
• queue.delete honors if-unused / if-empty, sends a server-initiated basic.cancel(consumerTag) to live consumers (on every node), and returns the residual count. Deleting a queue that does not exist replies delete-ok with a count of 0 and performs no side effects.
• if-empty is refused 406 rather than assumed satisfied when the count cannot be determined (the broker monitoring endpoint is unavailable) or while the connector still holds unacknowledged deliveries for the queue — that second case needs no outage at all, one manual-ack consumer with messages in flight is enough. Match on the reply text, and retry after the state settles. A queue that has never been published to is treated as empty.

Related