RabbitMQ Topology in Code: Exchange Types, Bindings, Quorum Queues and Streams

Version note: The code in this post was built and run with .NET 10, RabbitMQ 4.4 and RabbitMQ.Client 7.2.2.

Guess First

We have a topic exchange called bookstore.ordering with four queues bound to it. The binding keys: order.placed for billing.order-placed, order.placed again for notification.order-placed, order.* for analytics.order-status, and order.# for audit.orders. We publish four messages to the exchange, with routing keys order.placed, order.cancelled, order.shipped and order.shipped.partial. How many messages end up in each queue? How many copies are created in total?

And one more: billing.order-placed already exists as a quorum queue. What happens if another service declares the same name without arguments, that is, as a classic queue? Does it get an error, or does the queue's type change?

Exchange, Binding, Queue: The Path of a Message

In RabbitMQ a producer does not write to a queue; it writes to an exchange. The exchange looks at the message's routing key and its own type, and puts a copy of the message into each queue whose binding matches. If no binding matches, the message is dropped; in the previous post we caught that with the mandatory flag and PublishReturnException. The architectural side of queues, point-to-point and publish/subscribe lives in the Messaging and Event-Driven Architecture series post "Channels: Point-to-Point Queues, Publish/Subscribe Topics and Competing Consumers". Here we look at how that is built in RabbitMQ.

First, an exchange that is easy to miss: the default exchange, whose name is the empty string. Every queue is automatically bound to it with a routing key equal to the queue's own name. In the previous post's experiments, when we wrote BasicPublishAsync("", "lab.perf", body), we were sending the message straight to the lab.perf queue through this exchange. It is handy for quick experiments, but because it ties the producer to the consumer's queue name, use your own exchange in real flows.

Topic Exchange and Routing Key Design

We gather the topology for Bookstore's order events into a single class:

namespace Bookstore.Messaging;

public static class BookstoreTopology
{
    public const string Ordering = "bookstore.ordering";

    private static readonly Dictionary<string, object?> Quorum = new() { ["x-queue-type"] = "quorum" };

    public static async Task DeclareAsync(IChannel channel, CancellationToken ct = default)
    {
        await channel.ExchangeDeclareAsync(Ordering, ExchangeType.Topic, durable: true, cancellationToken: ct);

        await BindQueueAsync(channel, "billing.order-placed", "order.placed", ct);
        await BindQueueAsync(channel, "notification.order-placed", "order.placed", ct);
        await BindQueueAsync(channel, "analytics.order-status", "order.*", ct);
        await BindQueueAsync(channel, "audit.orders", "order.#", ct);
    }

    private static async Task BindQueueAsync(IChannel channel, string queue, string bindingKey, CancellationToken ct)
    {
        await channel.QueueDeclareAsync(queue, durable: true, exclusive: false, autoDelete: false,
            arguments: Quorum, cancellationToken: ct);
        await channel.QueueBindAsync(queue, Ordering, bindingKey, cancellationToken: ct);
    }
}

The output of rabbitmqctl list_queues name type messages after publishing the four messages:

audit.orders	quorum	4
notification.order-placed	quorum	1
billing.order-placed	quorum	1
analytics.order-status	quorum	3

Four messages produced nine copies. In a topic exchange the routing key is made of words separated by dots, and binding keys have two wildcards: * matches exactly one word, # matches zero or more words. order.shipped.partial has three words, so it does not match order.* but does match order.#. Billing and Notification have the same binding key, so each gets its own copy; this is what publish/subscribe looks like in RabbitMQ: every subscriber has its own queue.

Design the routing key like a namespace. In Bookstore the rule is <entity>.<event>[.<qualifier>]: order.placed, order.cancelled, order.shipped.partial. A few practical rules:

• Order from general to specific. order.placed.tr can be queried (order.placed.*, order.#); tr.placed.order becomes impossible to query.
• Do not put variable data in the routing key. If an order id or a customer e-mail ends up in the routing key, bindings become meaningless; that information belongs in the message body or a header.
• Put the version in the message's type property, not in the routing key. We will discuss contract versioning in the series' contract post.
• Start consumers with narrow bindings. A queue bound with # also receives every new event added later; except for consumers like audit that truly want everything, that is a surprise.

Direct, Fanout and Headers Exchanges

A direct exchange requires the routing key to be exactly equal to the binding key; no wildcards. It is a clean choice for command-style messages (like "create invoice" going to a single target). Because a topic exchange behaves like a direct exchange when no wildcards are used, many teams use topic for events from the start; if wildcards are needed later, they do not have to change the exchange.

A fanout exchange ignores the routing key completely and drops a copy into every bound queue. In Bookstore we use it for cache invalidation: when a book's price changes, every instance of the API (Application Programming Interface) must clear its own in-memory cache. Each instance opens a temporary queue for itself:

const string exchange = "bookstore.catalog.changed";
for (int instance = 1; instance <= 2; instance++)
{
    IChannel ch = await connection.CreateChannelAsync();
    await ch.ExchangeDeclareAsync(exchange, ExchangeType.Fanout, durable: true);
    // each instance opens its own temporary queue: server-named, deleted when the connection closes
    QueueDeclareOk q = await ch.QueueDeclareAsync(queue: "", durable: false, exclusive: true, autoDelete: true);
    await ch.QueueBindAsync(q.QueueName, exchange, routingKey: "");
    int id = instance;
    AsyncEventingBasicConsumer consumer = new(ch);
    consumer.ReceivedAsync += (_, ea) =>
    {
        Console.WriteLine($"instance-{id} ({q.QueueName}) got: {Encoding.UTF8.GetString(ea.Body.Span)}");
        return Task.CompletedTask;
    };
    await ch.BasicConsumeAsync(q.QueueName, autoAck: true, consumer);
}
await using IChannel pub = await connection.CreateChannelAsync();
await pub.BasicPublishAsync(exchange, "", Encoding.UTF8.GetBytes("book 42 price changed"));

instance-1 (amq.gen-XmdDueBrTfu8FFCsl84_fw) got: book 42 price changed
instance-2 (amq.gen-i58KEXRtTEK1gWU7Qsg22Q) got: book 42 price changed

For a queue declared with an empty name, the server assigns an amq.gen-... name. exclusive: true makes the queue belong to this connection only and deletes it when the connection closes. autoAck: true is deliberate here: if a cache-clear message is lost, at worst the cache corrects itself when its TTL (Time To Live) expires, and if the instance dies there is no cache left to clear anyway. These queues cannot be quorum queues; quorum queues do not support being exclusive and temporary, so this is a job for classic queues.

A headers exchange looks at message headers instead of the routing key. In the binding, x-match selects "all must match" (all) or "any may match" (any):

await ch.QueueBindAsync("shipping.express-eu", exchange, routingKey: "",
    arguments: new Dictionary<string, object?> { ["x-match"] = "all", ["region"] = "eu", ["speed"] = "express" });

foreach ((string region, string speed) in new[] { ("eu", "express"), ("eu", "standard"), ("tr", "express") })
{
    BasicProperties props = new() { Headers = new Dictionary<string, object?> { ["region"] = region, ["speed"] = speed } };
    await ch.BasicPublishAsync(exchange, "", mandatory: false, props, Encoding.UTF8.GetBytes($"{region}/{speed}"));
}

shipping.express-eu	quorum	1

Of three messages only eu/express arrived. A headers exchange looks flexible but is rarely needed in practice: doing the same thing with a routing key like shipping.express.eu and a topic exchange is both faster and easier to read in the management UI (user interface).

Declaring the Topology Idempotently

Each service declares the exchanges and queues it uses at startup. The producer declares the exchange; the consumer declares the exchange, its own queue and its binding. That makes the order in which services start irrelevant. Declarations are idempotent, with one condition: the same name must come with the same arguments. The answer to the second question at the top:

// same name, different arguments: try to declare it as a classic queue
await ch.QueueDeclareAsync("billing.order-placed", durable: true, exclusive: false, autoDelete: false);

The AMQP operation was interrupted: AMQP close-reason, initiated by Peer, code=406, text='PRECONDITION_FAILED - inequivalent arg 'x-queue-type' for queue 'billing.order-placed' in vhost '/': received 'classic' but current is 'quorum'', classId=50, methodId=10
Channel open after error: False

The queue's type did not change; the declaring channel was closed. That is a good thing: if two services disagree about the same queue, you find out at startup, loudly. But there are two consequences. First, channel-level errors kill the channel; declare topology on a separate, short-lived channel and do not put your publish channel at risk. Second, queue arguments cannot be changed. To change a queue's type or its x-max-length argument, you must delete and recreate the queue, which means moving its messages somewhere else. That is why settings that may change belong in a policy rather than in arguments; we will get there shortly.

Queue Types: Classic, Quorum and Stream

RabbitMQ 4.x has three queue types, and they differ in behaviour, not just performance:

The default choice for durable business messages is the quorum queue. In a single-node development environment it does not replicate, but the code and behaviour stay the same as in a production cluster. delivery-limit is particularly valuable: if a message is delivered 20 times and nacked every time, the quorum queue drops it, or sends it to the DLX (Dead Letter Exchange) if dead-lettering is configured. In a classic queue the same message loops forever. We will connect this limit to the retry strategy in the series' retry post.

A stream is a log: messages stay after they are acked, and are deleted segment by segment when retention (x-max-age, x-max-length-bytes) is reached. Each consumer reads from its own offset. A stream for audit:

await ch.QueueDeclareAsync("audit.order-stream", durable: true, exclusive: false, autoDelete: false,
    arguments: new Dictionary<string, object?> { ["x-queue-type"] = "stream", ["x-max-age"] = "7D" });

for (int reader = 1; reader <= 2; reader++)
{
    await using IChannel rc = await connection.CreateChannelAsync();
    // prefetch is mandatory for a stream consumer
    await rc.BasicQosAsync(0, 100, false);
    List<string> seen = [];
    AsyncEventingBasicConsumer consumer = new(rc);
    consumer.ReceivedAsync += async (_, ea) =>
    {
        seen.Add(Encoding.UTF8.GetString(ea.Body.Span));
        await rc.BasicAckAsync(ea.DeliveryTag, false);
    };
    await rc.BasicConsumeAsync("audit.order-stream", autoAck: false, consumerTag: "", noLocal: false,
        exclusive: false, arguments: new Dictionary<string, object?> { ["x-stream-offset"] = "first" },
        consumer: consumer);
    await Task.Delay(700);
    Console.WriteLine($"reader-{reader} read {seen.Count}: {string.Join(", ", seen)}");
}

Published 5 messages to the stream
reader-1 read 5: order-1, order-2, order-3, order-4, order-5
reader-2 read 5: order-1, order-2, order-3, order-4, order-5

The second reader read from the beginning the messages the first one had acked. x-stream-offset accepts first, last, next, a numeric offset or a timestamp. Reading a stream over AMQP (Advanced Message Queuing Protocol) 0-9-1 requires prefetch and manual acks; here an ack does not delete the message, it only grants credit for flow control. If you need high throughput, the RabbitMQ documentation recommends the stream-specific binary protocol and its clients (RabbitMQ.Stream.Client for .NET). If replay is the dominant need, Kafka is also on the table; we compared them in the first post of the series.

TTL, Max-Length and Overflow

Queues cannot grow forever. When a consumer is down for hours, you do not want a queue to fill the broker's memory and disk. There are two main tools: giving messages a lifetime (x-message-ttl, or the per-message Expiration property) and putting a size limit on the queue (x-max-length, x-max-length-bytes). x-overflow decides what happens when the limit is reached: the default drop-head discards the oldest message, reject-publish rejects the new one. For messages like orders where loss is unacceptable, telling the producer "stop" is more honest than silently discarding the oldest:

await ch.QueueDeclareAsync("lab.limited", durable: true, exclusive: false, autoDelete: false,
    arguments: new Dictionary<string, object?>
    {
        ["x-queue-type"] = "quorum",
        ["x-max-length"] = 3,
        ["x-overflow"] = "reject-publish"
    });

m1: confirmed
m2: confirmed
m3: confirmed
m4: confirmed
m5: PublishException IsReturn=False
m6: PublishException IsReturn=False

Two details. First, you only see the rejection if publisher confirmations are on; the broker sends a basic.nack and the client turns it into a PublishException with IsReturn=False. With a producer that has confirmations off, the message is silently lost. Second, in our experiment the quorum queue did not apply the limit exactly at the boundary but one message late: the fourth message was accepted, the fifth was rejected. Think of limits as a safety belt, not an exact number.

Client or Policy?

Should a setting be given in code as a queue argument, or defined on the broker as a policy? A policy is a set of settings applied to queues whose names match a pattern (regex) and can be changed at runtime:

docker exec mqwork1-rabbit rabbitmqctl set_policy --apply-to quorum_queues --priority 10 billing-limits '^billing\.' '{"delivery-limit":5,"max-length":100000,"overflow":"reject-publish"}'
docker exec mqwork1-rabbit rabbitmqctl list_queues name policy effective_policy_definition arguments

name	policy	effective_policy_definition	arguments
audit.orders		#{}	[{"x-queue-type","quorum"}]
billing.order-placed	billing-limits	[{<<"delivery-limit">>, 5}, {<<"max-length">>, 100000}, {<<"overflow">>, reject-publish}]	[{"x-queue-type","quorum"}]

The queue picked up the new limits without being redeclared. The practical split:

• In code (arguments): what defines the queue's identity and is required for the application to be correct. Queue type (x-queue-type), durable, exclusive, the fact that it is a stream.
• In a policy: what varies by environment and may need tuning at runtime. TTL, max-length, overflow, delivery-limit, dead-letter exchange.

Know the precedence rules. According to the RabbitMQ documentation, when the same key is in both an argument and a user policy, the argument generally wins; for max-length, the smaller of the two applies. Only one (user) policy applies to a queue at a time; of two policies with overlapping patterns, the one with the higher priority wins. There are also operator policies, so operators can set limits that cannot be overridden. So setting x-max-length in code and later trying to raise it with a policy does not work. Leave settings that will change to a policy from the start.

Production Notes

Have an answer to the question of who declares the topology. In small systems "each service declares what it needs" is enough. In large systems, managing topology with a definitions file (rabbitmqctl export_definitions) or infrastructure code and restricting applications' declare permissions prevents queues being created with the wrong arguments.

Watch the number of queues. Every quorum queue is a separate Raft group; thousands of quorum queues put serious load on a cluster. Instead of a queue per customer or per order, separate traffic with routing keys and bindings.

Watch queues bound with #. A queue that receives everything, like audit.orders, becomes the fastest-growing queue when its consumer stops. Always give it a retention limit, such as max-length or a stream.

Design dead-lettering together with the topology. When a quorum queue's delivery-limit is reached and no DLX is defined, the message is silently deleted. We will build how to manage the DLX and the DLQ (Dead Letter Queue) in detail in the series' dead-letter post; for now, make sure the policy has a dead-letter-exchange.

Common Pitfalls

Using classic queues as "the default". If you pass no arguments to QueueDeclareAsync you get a classic queue; in 4.x that means an unreplicated queue. For durable business messages, write x-queue-type explicitly.

Trying to change a queue argument later. You get PRECONDITION_FAILED and the channel closes. Move everything that may change into a policy.

Putting variable data in the routing key. Keys like order.placed.01a1b8de... make bindings meaningless and become unreadable in the management UI.

Trying quorum queues when fanning out to every instance. Exclusive, temporary queues have to be classic.

Consuming a stream like a queue. An ack does not delete the message; if you do not store the offset, the consumer either reprocesses everything with first on every restart or skips the messages in between with next.

Using reject-publish without confirmations. The broker rejects, the producer never hears about it, and the message is lost.

A Small Experiment

★☆☆ Publish a message with the routing key order.placed.tr. Which of the four queues above receive it? Guess first, then check with rabbitmqctl list_queues.

★★☆ Define a policy for billing.order-placed containing message-ttl and dead-letter-exchange. Publish a message while the consumer is stopped, and watch where the message goes when the TTL expires.

★★★ Write a consumer that reads audit.order-stream and stores the last offset it processed (from the message's x-stream-offset header) in a file. When the process restarts it should continue from that offset; show that no message is processed twice.

We have seen how to build RabbitMQ topology with code and policies. In the next post we move to a managed broker: queues, topics and subscriptions with the Azure Service Bus .NET SDK (Software Development Kit), ServiceBusProcessor, PeekLock and lock duration, ordered processing with sessions, and trying all of it locally with the Service Bus emulator.