Consuming from RabbitMQ

This content was generated with the assistance of AI and should be verified against the official documentation before being relied on in production; see the section bibliography for the reference material consulted while preparing these pages.

Publishing to RabbitMQ covered getting a message routed into a queue reliably; this page covers the other side — reading it back out, deciding when it counts as handled, and controlling how many deliveries a consumer takes on at once. It covers the AMQP 0-9-1 consumer API end to end (push versus pull, acknowledgement modes, prefetch, priorities, single active consumer, exclusivity, cancellation), then the equivalent for streams, and ends with a complete manual-acknowledgement consumer example against the RabbitMQ Java client.

Push (basic.consume) vs. pull (basic.get)

RabbitMQ offers two fundamentally different ways to read a message off a queue:

  • Push — channel.basicConsume(queue, autoAck, deliverCallback, cancelCallback) registers a long-lived subscription; the broker pushes each new message to the registered callback as it becomes available, on a connection thread the client library manages. This is the normal, recommended way to consume — every example elsewhere on this page uses it.

  • Pull — channel.basicGet(queue, autoAck) polls the queue once and returns a single GetResponse (or null if the queue was empty) synchronously. RabbitMQ’s own documentation calls repeated polling with basic.get "highly inefficient" compared to a standing consumer, and discourages it outside of one-off inspection or debugging.

GetResponse response = channel.basicGet("orders", false); // autoAck = false
if (response != null) {
    System.out.println(new String(response.getBody()));
    channel.basicAck(response.getEnvelope().getDeliveryTag(), false);
} // a null response just means the queue was empty at that instant -- not an error

Acknowledgement modes

An acknowledgement is how a consumer tells the broker a delivery is done with — safe to remove from the queue. Until it is acknowledged, the broker keeps the delivery around as unacked, ready to redeliver if the consumer’s channel or connection closes before acknowledging.

Automatic acknowledgement

With autoAck=true (the boolean passed to basicConsume/basicGet), the broker considers a message acknowledged the instant it is written to the network, before the consumer’s callback has even run. This is the simplest mode and the fastest, but it gives up all delivery safety: a consumer that crashes mid-processing loses that message permanently, since the broker already considers it delivered and removes it. Reach for automatic acknowledgement only where an occasional lost message is genuinely acceptable.

Manual acknowledgement: basic.ack, basic.nack, basic.reject

With autoAck=false, the application acknowledges (or rejects) explicitly, once processing has actually finished:

Method Meaning

channel.basicAck(deliveryTag, multiple)

Positively acknowledges — processing succeeded, the broker may remove the message(s). multiple=true acknowledges every unacked delivery on this channel up to and including deliveryTag, not just that one.

channel.basicNack(deliveryTag, multiple, requeue)

Negatively acknowledges one or more deliveries (the only ack method that also supports multiple), with requeue deciding what happens next (below).

channel.basicReject(deliveryTag, requeue)

Negatively acknowledges exactly one delivery — the single-message equivalent of basicNack, with no multiple flag.

requeue=true puts the message back on the queue for redelivery (to this consumer or another one, immediately, which can cause a tight redelivery loop for a message that always fails the same way); requeue=false instead drops the message, or — if the queue is configured with a dead-letter exchange — dead-letters it there instead of dropping it outright.

DeliverCallback deliverCallback = (consumerTag, delivery) -> {
    try {
        process(delivery.getBody());
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    } catch (RecoverableException e) {
        // transient failure -- try again, but not immediately from this same consumer
        channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
    } catch (Exception e) {
        // not recoverable by retrying -- drop it (or let a configured DLX catch it) rather than loop forever
        channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false);
    }
};

Prefetch (QoS)

channel.basicQos(prefetchCount) caps how many unacknowledged deliveries the broker will have outstanding to a consumer at once — once that many are unacked, the broker stops pushing more until at least one is acknowledged. Unlike the AMQP 0-9-1 specification, which defines prefetch as shared across every consumer on a channel, RabbitMQ applies a plain basicQos(prefetchCount) call separately to each new consumer registered on that channel afterward. Passing global=true (basicQos(prefetchCount, true)) switches to a single limit shared across every consumer on the channel instead, matching the spec’s own semantics.

A prefetch of 0 means unlimited — the broker pushes as fast as the network allows, which risks one slow consumer accumulating an unbounded number of unacked (and therefore un-redeliverable-elsewhere) messages. A small prefetch (RabbitMQ’s own guidance suggests starting around 100-300 and tuning from there) trades some throughput for fairer distribution across multiple consumers and a bounded amount of in-flight, unacknowledged work per consumer.

channel.basicQos(50); // this channel's next basicConsume gets its own 50-message cap

See RabbitMQ Documentation — Consumer Prefetch for the full per-channel vs. per-consumer vs. global semantics and tuning guidance.

Delivery acknowledgement timeout

consumer_timeout bounds how long a delivery may sit unacknowledged before RabbitMQ gives up on the consumer holding it: by default 30 minutes, evaluated at roughly one-minute intervals. A consumer that exceeds it has its channel closed with a PRECONDITION_FAILED error, and every delivery outstanding on that channel is requeued as if the consumer had disappeared. It exists to protect the broker (and disk space, on a durable queue) from a stuck or hung consumer holding deliveries open indefinitely rather than crashing or requeuing cleanly — it is not meant to be tuned down to double as an application-level per-message processing deadline; RabbitMQ’s own guidance discourages values under five minutes.

Consumer priorities

Multiple consumers on the same queue normally receive messages round-robin. Setting x-priority (an integer, default 0, negative values allowed) in a consumer’s basicConsume arguments changes that: the broker prefers delivering to the highest-priority active consumer(s) first, falling back to a lower-priority one only once every higher-priority consumer is unable to accept more (blocked by its own prefetch limit, for example). Consumers tied at the same highest priority still share deliveries round-robin among themselves.

Map<String, Object> consumeArguments = Map.of("x-priority", 10);
channel.basicConsume("orders", false, consumeArguments, deliverCallback, cancelCallback);

Single active consumer

Single active consumer (x-single-active-consumer, set true as a queue argument at declaration time) limits a queue to delivering to exactly one of its registered consumers at a time — every other registered consumer sits idle, ready to be promoted the instant the active one’s channel or connection closes. It is how a queue gets both several standby consumer instances (for fast failover) and strictly ordered, non-competing processing, which a plain prefetch-limited multi-consumer setup cannot guarantee since deliveries can be spread and processed out of order across consumers. Both quorum and classic queues support it; RabbitMQ’s own streams use the closely related concept per partition of a super stream, covered on RabbitMQ Queue Types.

Map<String, Object> queueArguments = Map.of("x-single-active-consumer", true);
channel.queueDeclare("orders", true, false, false, queueArguments);

Exclusive consumers

The exclusive flag on basicConsume requests this consumer be the only one ever registered on the queue — the call fails outright if another consumer is already attached, rather than both being allowed to coexist. It applies to classic queues only: a quorum queue silently ignores the flag, since single active consumer (above) already covers quorum queues' equivalent need with graceful failover instead of an outright registration failure.

boolean exclusive = true;
channel.basicConsume("orders", false, "", false, exclusive, null, deliverCallback, cancelCallback);

Consumer cancellation

channel.basicCancel(consumerTag) unsubscribes a consumer explicitly — no further deliveries are dispatched to it afterward, though deliveries already in flight are unaffected. A consumer can also be cancelled by RabbitMQ itself, without the application asking for it — most commonly because the queue it was consuming from was deleted. The cancelCallback passed to basicConsume is invoked in that case, which is why a real consumer always registers one instead of only a deliverCallback, to notice and react to an involuntary cancellation instead of silently stopping.

CancelCallback cancelCallback = consumerTag ->
        System.out.println("consumer " + consumerTag + " was cancelled by the broker");

String consumerTag = channel.basicConsume("orders", false, deliverCallback, cancelCallback);
// ...
channel.basicCancel(consumerTag); // explicit, application-initiated cancellation

Direct reply-to

Publishing to RabbitMQ introduced correlationId/replyTo as the pair an RPC-style publisher uses to route a reply back to itself. The straightforward way to do that is a dedicated, exclusive reply queue declared per request — correct, but relatively expensive at scale, since every request-carrying connection ends up declaring and tearing down its own queue. Direct reply-to is a pseudo-queue built into RabbitMQ specifically for this pattern: a client sets replyTo to the special value amq.rabbitmq.reply-to and consumes from that same pseudo-queue name, and a reply published back to it is routed directly to the requester without ever being durably stored, without a metadata-store entry, and without the per-request queue declaration/deletion churn.

// requester side
String correlationId = UUID.randomUUID().toString();
channel.basicConsume("amq.rabbitmq.reply-to", true, (consumerTag, delivery) -> {
    if (correlationId.equals(delivery.getProperties().getCorrelationId())) {
        System.out.println("reply: " + new String(delivery.getBody()));
    }
}, consumerTag -> { });

AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
        .correlationId(correlationId)
        .replyTo("amq.rabbitmq.reply-to")
        .build();
channel.basicPublish("", "rpc.orders", props, "{\"orderId\":\"order-42\"}".getBytes());

Direct reply-to trades away durability and at-least-once delivery — a reply published while no requester is consuming from amq.rabbitmq.reply-to is simply lost, not buffered — which is the right trade for a low-latency RPC reply that is only ever useful for a few seconds after the request, and the wrong one for anything that needs a durably buffered response.

stateDiagram-v2 [*] --> Delivered: broker dispatches (push or basic.get) Delivered --> Acked: basic.ack Delivered --> RequeuedForRedelivery: basic.nack / basic.reject (requeue=true) Delivered --> DeadLetteredOrDropped: basic.nack / basic.reject (requeue=false) Delivered --> RequeuedForRedelivery: consumer_timeout exceeded (channel closed) RequeuedForRedelivery --> Delivered: redelivered (to this or another consumer) Acked --> [*] DeadLetteredOrDropped --> [*]

Consuming from streams

Reading a stream is non-destructive — consuming never removes a message, so where a subscription starts is a choice the consumer makes explicitly, not a broker-side default. Through the stream protocol’s com.rabbitmq.stream:stream-client library, that choice is an OffsetSpecification passed when building a Consumer:

OffsetSpecification Starting point

.first()

The first offset the stream still retains — a full replay of whatever hasn’t been trimmed yet.

.last()

The last written chunk — roughly "recent messages," not a single exact message.

.next()

The next offset to be written — only messages produced from this point on, skipping everything already stored.

.offset(long)

An exact, previously-known offset — resuming from a specific position, for example one read back from external storage.

.timestamp(long)

The first offset at or after the given timestamp.

 
Consumer consumer = environment.consumerBuilder()
        .stream("orders-stream")
        .name("orders-consumer")           // a stable name enables the offset tracking below
        .offset(OffsetSpecification.next())
        .messageHandler((context, message) -> process(message))
        .build();

Offset tracking

Because a stream keeps every retained message regardless of consumption, offset tracking — remembering how far a given named consumer has read — is an explicit, opt-in feature rather than an automatic broker-side commit. Automatic tracking (autoTrackingStrategy()) periodically stores the current offset on the consumer’s behalf, tunable by a message-count threshold and a flush interval; manual tracking (manualTrackingStrategy()) instead leaves it entirely to the application, which calls context.storeOffset() from inside the message handler once it decides a given message’s processing (and everything before it) is durably finished — the stream analogue of a manual commitSync() on a Kafka consumer (see Kafka Consumers). A consumer that reconnects with the same name and no explicit OffsetSpecification resumes from its own last-tracked offset automatically.

See RabbitMQ Documentation — Streams and Superstreams (Partitioned Streams) and the RabbitMQ Stream Java client reference linked from Publishing to RabbitMQ for the complete consumer-side flow control, credit and tracking-strategy configuration.

A plain-Java consumer example with manual acks

The example below combines manual acknowledgement, a bounded prefetch, and a cancel callback into one realistic consumer against com.rabbitmq.client:amqp-client.

package com.example.messaging.rabbitmq;

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

public class ManualAckOrderConsumer {

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        // bound how many unacked deliveries this consumer can be holding at once
        channel.basicQos(50);

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            try {
                process(delivery.getBody());
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            } catch (Exception e) {
                // requeue=false -- a configured dead-letter exchange catches it instead of retrying forever
                channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false);
            }
        };

        CancelCallback cancelCallback = consumerTag ->
                System.err.println("consumer " + consumerTag + " cancelled by the broker (queue deleted?)");

        String consumerTag = channel.basicConsume("orders", false, deliverCallback, cancelCallback);

        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            try {
                channel.basicCancel(consumerTag);
                connection.close();
            } catch (Exception e) {
                // best-effort shutdown
            }
        }));
    }

    private static void process(byte[] body) {
        // apply the order; an exception here causes a nack(requeue=false) above
    }
}

See RabbitMQ Documentation — Consumers for the full push/pull, acknowledgement and cancellation API this example builds on. For the publishing side of this same queue, see Publishing to RabbitMQ.

References