Publishing to 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.

RabbitMQ Architecture and the AMQP Model covered what an exchange, binding and queue are; this page covers the publishing API itself — how a client actually gets a message routed reliably, how it finds out when routing or delivery failed, and how it publishes into a stream instead of a classic or quorum queue. It works through plain AMQP 0-9-1 publishing first (basicPublish, the mandatory flag, publisher confirms, message properties), then the stream-specific publishing API, and ends with a complete, runnable confirmed-publishing example against the RabbitMQ Java client.

Publishing to an exchange

A publisher never targets a queue directly — channel.basicPublish(exchange, routingKey, properties, body) always names an exchange (the empty string for the default exchange) and a routing key, and it is the exchange’s bindings that decide which queue, if any, actually receives a copy. basicPublish itself is asynchronous and returns immediately once the frame has been written to the client’s internal buffer — it does not wait for the broker to accept, route, or store anything, which is exactly why the mechanisms below exist.

The `mandatory` flag and returned messages

By default, a message an exchange cannot route to any queue is simply dropped, silently, with no error raised anywhere. Setting the mandatory flag on basicPublish changes that: if the message cannot be routed, the broker sends it back to the publisher instead of dropping it, as a basic.return — delivered to a return listener registered on the same channel, never as an exception thrown out of basicPublish itself.

channel.addReturnListener(returnMessage -> {
    // invoked once per unroutable message published with mandatory = true on this channel
    System.out.printf("returned: exchange=%s routingKey=%s replyCode=%d replyText=%s%n",
            returnMessage.getExchange(), returnMessage.getRoutingKey(),
            returnMessage.getReplyCode(), returnMessage.getReplyText());
});

boolean mandatory = true;
channel.basicPublish("orders", "orders.unknown-region", mandatory, null,
        "{\"orderId\":\"order-42\"}".getBytes());

mandatory costs a small amount of broker-side bookkeeping per publish, so it is normally reserved for exchanges and routing keys where an unroutable message is a real, actionable failure — not turned on globally for traffic where a silently dropped message is acceptable.

Alternate exchanges

An alternate exchange is the per-exchange equivalent of mandatory, configured once at declaration time instead of per message: any message the primary exchange cannot route to a single queue is republished, unchanged, to the configured alternate exchange, which can itself route it anywhere — typically to a catch-all queue used for monitoring or later replay.

Map<String, Object> exchangeArguments = Map.of("alternate-exchange", "orders.unrouted");

channel.exchangeDeclare("orders.unrouted", BuiltinExchangeType.FANOUT, true);
channel.queueDeclare("orders.unrouted.queue", true, false, false, null);
channel.queueBind("orders.unrouted.queue", "orders.unrouted", "");

channel.exchangeDeclare("orders", BuiltinExchangeType.TOPIC, true, false, exchangeArguments);

mandatory and an alternate exchange are not mutually exclusive, but only one of them actually fires for a given unroutable message: if an alternate exchange is configured and itself routes the message successfully, the message is not considered unroutable any more, so no basic.return is sent even if mandatory was also set.

Persistent delivery mode

deliveryMode on AMQP.BasicProperties controls whether the message itself survives a broker restart, entirely independently of whether the queue it lands in is durable (see RabbitMQ Architecture and the AMQP Model for the queue-level durable property):

deliveryMode Meaning

`deliveryMode`	Meaning
`1` (transient, the default)	Kept in memory only; lost on a broker restart even if the queue it sits in is durable.
`2` (persistent)	Written to disk before being acknowledged as routed; survives a broker restart, at the cost of the extra disk write on the publishing path.

1 (transient, the default)

Kept in memory only; lost on a broker restart even if the queue it sits in is durable.

2 (persistent)

Written to disk before being acknowledged as routed; survives a broker restart, at the cost of the extra disk write on the publishing path.

A message needs both deliveryMode=2 and a durable queue to actually survive a restart — persistent delivery mode into a non-durable queue is pointless (the queue itself disappears on restart), and a durable queue holding transient messages loses exactly those messages, not the queue’s definition.

Publisher confirms

Publisher confirms are RabbitMQ’s answer to basicPublish’s fire-and-forget nature: an extension the client opts into per channel with `channel.confirmSelect(), after which the broker acknowledges each published message once it has been handled — persisted to disk for a persistent message on a durable queue, or simply enqueued for a transient one — with the acknowledgement correlated back to the publish by a channel-scoped, strictly increasing delivery tag (channel.getNextPublishSeqNo() returns the tag the next publish will receive). A negative acknowledgement (basic.nack) is rare — it signals an internal broker error, not a routing failure, which is what mandatory/returns and alternate exchanges handle instead.

Three strategies use confirms, trading latency for throughput:

Strategy How it works

Strategy	How it works
Individual (synchronous)	Publish one message, then call `channel.waitForConfirms()` (or `waitForConfirmsOrDie()`, which throws on a `nack`) before publishing the next. Simplest and safest, but throughput is capped by one broker round trip per message.
Batched	Publish a batch of messages back to back, then call `waitForConfirmsOrDie()` once for the whole batch. Far higher throughput than individual confirms, at the cost of not knowing which message in the batch failed if the wait throws — a failure means redelivering (or re-checking) the entire batch.
Asynchronous	Register a confirm listener (`channel.addConfirmListener(ackCallback, nackCallback)`) and keep publishing without ever blocking on confirms; the client tracks which delivery tags are still outstanding itself and the listener removes them as acknowledgements arrive, out of band from the publishing thread. Highest throughput and the strategy RabbitMQ’s own documentation recommends for anything sending a non-trivial volume of messages.

Individual (synchronous)

Publish one message, then call channel.waitForConfirms() (or waitForConfirmsOrDie(), which throws on a nack) before publishing the next. Simplest and safest, but throughput is capped by one broker round trip per message.

Batched

Publish a batch of messages back to back, then call waitForConfirmsOrDie() once for the whole batch. Far higher throughput than individual confirms, at the cost of not knowing which message in the batch failed if the wait throws — a failure means redelivering (or re-checking) the entire batch.

Asynchronous

Register a confirm listener (channel.addConfirmListener(ackCallback, nackCallback)) and keep publishing without ever blocking on confirms; the client tracks which delivery tags are still outstanding itself and the listener removes them as acknowledgements arrive, out of band from the publishing thread. Highest throughput and the strategy RabbitMQ’s own documentation recommends for anything sending a non-trivial volume of messages.

channel.confirmSelect(); // opt this channel into publisher confirms

ConcurrentNavigableMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();

channel.addConfirmListener((deliveryTag, multiple) -> {
    // ack: everything up to and including deliveryTag succeeded when multiple == true
    if (multiple) {
        outstandingConfirms.headMap(deliveryTag, true).clear();
    } else {
        outstandingConfirms.remove(deliveryTag);
    }
}, (deliveryTag, multiple) -> {
    // nack: the broker could not handle this message (or these messages, if multiple) -- log and/or retry
    ConcurrentNavigableMap<Long, String> nacked = multiple
            ? outstandingConfirms.headMap(deliveryTag, true)
            : outstandingConfirms.tailMap(deliveryTag, true).headMap(deliveryTag, true);
    System.err.println("nacked: " + nacked);
    outstandingConfirms.remove(deliveryTag);
});

String body = "{\"orderId\":\"order-42\"}";
outstandingConfirms.put(channel.getNextPublishSeqNo(), body);
channel.basicPublish("orders", "orders.eu.created", null, body.getBytes());

Acknowledgements can arrive in a different order than the messages that triggered them, and multiple=true acknowledges (or negatively acknowledges) every outstanding delivery tag up to and including the one given, not just that single tag — both callbacks above account for that instead of assuming one confirm per message.

sequenceDiagram participant P as Publisher participant B as Broker P->>B: channel.confirmSelect() Note over P: publish A (seq 1) and B (seq 2) P->>B: basic.publish (seq 1, routing key "orders.eu.created") P->>B: basic.publish (seq 2, routing key "orders.unknown-region", mandatory=true) Note over B: seq 1 routes to a queue -- seq 2 matches no binding B-->>P: basic.return (seq 2 -- unroutable) B-->>P: basic.ack (seq 1) Note over P: seq 2 is not routed but is still handled (returned), so it is later acked too B-->>P: basic.ack (seq 2)

See RabbitMQ Documentation — Consumer Acknowledgements and Publisher Confirms for the full acknowledgement/negative-acknowledgement contract, and RabbitMQ Documentation — Reliability Guide for how confirms fit into a publisher’s overall recovery strategy, including retransmitting on a channel or connection failure.

Message properties and headers

RabbitMQ Architecture and the AMQP Model already introduced AMQP.BasicProperties and deliveryMode; from the publishing side, two properties matter enough to call out specifically. headers is an open, string-keyed Map<String, Object> a publisher attaches freely — it is what a headers exchange’s x-match binding matches against, and otherwise arbitrary metadata a consumer can inspect without deserializing the payload (a trace ID, a schema version, a tenant identifier). correlationId and replyTo together are what an RPC-style publisher uses to get a reply routed back to itself without a dedicated reply queue per request — covered in full, from the consuming side, on Consuming from RabbitMQ's direct reply-to section.

AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
        .contentType("application/json")
        .deliveryMode(2)
        .correlationId(UUID.randomUUID().toString())
        .headers(Map.of("trace-id", "4c2f9e", "schema-version", 2))
        .build();

channel.basicPublish("orders", "orders.eu.created", properties,
        "{\"orderId\":\"order-42\"}".getBytes());

See RabbitMQ Documentation — Publishers for the complete AMQP.BasicProperties reference table.

Publishing to streams

RabbitMQ Queue Types introduced streams as an append-only, replicated log. Publishing into one uses a dedicated, higher-throughput binary protocol (the stream protocol, port 5552 by default) rather than AMQP 0-9-1, through the separate com.rabbitmq.stream:stream-client library — an Environment and a Producer, not a Connection/Channel.

Environment environment = Environment.builder()
        .host("localhost")
        .build();

Producer producer = environment.producerBuilder()
        .stream("orders-stream")
        .name("orders-producer")   // a stable producer name is required for deduplication below
        .build();

long publishingId = producer.getLastPublishingId() + 1; // resume after the last confirmed id on restart

producer.send(
        producer.messageBuilder()
                .publishingId(publishingId)
                .addData("{\"orderId\":\"order-42\"}".getBytes())
                .build(),
        confirmationStatus -> {
            if (!confirmationStatus.isConfirmed()) {
                System.err.println("publish not confirmed: " + confirmationStatus.getMessage());
            }
        });

Publishing IDs and deduplication

A publishing ID is a strictly increasing, application-assigned sequence number attached to each message through messageBuilder().publishingId(long). Combined with a stable producer name (.name("orders-producer") above), it lets the broker deduplicate: republishing a publishing ID it has already seen from that named producer is silently ignored rather than stored a second time. producer.getLastPublishingId() returns the last publishing ID the broker confirmed for this producer name, which is exactly what a producer recovering after a crash or restart uses to resume its sequence without either skipping or repeating IDs. This mechanism requires a single producer instance per name, publishing from a single thread — two instances sharing a name, or one instance publishing concurrently from multiple threads, break the strictly-increasing sequence the deduplication relies on.

Flow control and blocked connections

On plain AMQP 0-9-1 connections (including a stream’s non-stream-protocol traffic and every other queue type on this page), RabbitMQ protects itself against a publisher outrunning the broker’s ability to keep up by blocking the connection — refusing to read further data from it — once a resource alarm (memory or disk) trips, and notifying the client with a connection.blocked method so well-behaved clients can react (pause publishing, surface backpressure) rather than simply stalling with no explanation:

connection.addBlockedListener(
        reason -> System.out.println("connection blocked: " + reason),
        () -> System.out.println("connection unblocked"));

The stream protocol instead applies flow control natively per connection, without needing the blocked-connection extension: a stream Producer caps how many unconfirmed messages it will hold at once (maxUnconfirmedMessages, 10,000 by default) and Producer#send simply starts blocking the calling thread once that limit is reached, resuming only as confirmations free up room — a direct form of backpressure rather than a broker-initiated pause.

See RabbitMQ Documentation — Flow Control and RabbitMQ Documentation — Blocked Connection Notifications for the AMQP 0-9-1 mechanism’s full alarm thresholds and client-library support, and the RabbitMQ Stream Java client reference below for the stream protocol’s own producer/consumer flow-control settings.

Batching for throughput

The stream Producer batches outgoing messages itself, on top of whatever batching an application does: batchSize (default 100) bounds how many messages accumulate before a batch is sent, and sub-entry batching (subEntrySize, optionally combined with compression(Compression.ZSTD) or another supported codec) goes further by packing several messages into the storage slot the stream would otherwise use for one, trading a small amount of added latency and CPU for meaningfully higher throughput on high-volume streams. Both settings are configured on the same producerBuilder() chain used above.

See RabbitMQ Stream Java Client Documentation Guide for the complete producer/consumer API, including every batching, compression and flow-control setting.

A plain-Java publishing example with confirms

The example below combines asynchronous publisher confirms, the mandatory flag with a return listener, and persistent delivery mode into one realistic publisher against com.rabbitmq.client:amqp-client.

package com.example.messaging.rabbitmq;

import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class ConfirmedOrderPublisher {

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

        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            channel.exchangeDeclare("orders", BuiltinExchangeType.TOPIC, true);
            channel.confirmSelect();

            ConcurrentNavigableMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();

            channel.addReturnListener(returnMessage -> System.err.printf(
                    "unroutable: routingKey=%s replyText=%s%n",
                    returnMessage.getRoutingKey(), returnMessage.getReplyText()));

            channel.addConfirmListener((deliveryTag, multiple) -> {
                if (multiple) {
                    outstandingConfirms.headMap(deliveryTag, true).clear();
                } else {
                    outstandingConfirms.remove(deliveryTag);
                }
            }, (deliveryTag, multiple) -> {
                String body = outstandingConfirms.get(deliveryTag);
                System.err.println("nacked, will be retried by an outer loop: " + body);
                outstandingConfirms.remove(deliveryTag);
            });

            AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
                    .contentType("application/json")
                    .deliveryMode(2) // persistent -- survives a broker restart on a durable queue
                    .correlationId(UUID.randomUUID().toString())
                    .headers(Map.of("trace-id", "4c2f9e"))
                    .build();

            for (int i = 0; i < 10; i++) {
                String body = "{\"orderId\":\"order-" + i + "\"}";
                boolean mandatory = true;
                outstandingConfirms.put(channel.getNextPublishSeqNo(), body);
                channel.basicPublish("orders", "orders.eu.created", mandatory, properties, body.getBytes());
            }

            // in a real application this loop lives on a scheduled check rather than a fixed sleep -- shown
            // here only to give outstanding confirms time to arrive before the connection closes
            Thread.sleep(2_000);
            if (!outstandingConfirms.isEmpty()) {
                System.err.println("still outstanding at shutdown: " + outstandingConfirms.keySet());
            }
        }
    }
}

See RabbitMQ Documentation — Publishers for the mandatory flag and message-properties reference this example builds on, and RabbitMQ Documentation — Java Client API Guide for the Channel/Connection API and its automatic topology and connection recovery behavior. For the consuming side of this same exchange and queue, see Consuming from RabbitMQ.