RabbitMQ

The problem: slow work after placing an order

In our online shop, after an order is placed, a receipt email must be sent. The email server sometimes takes a few seconds, and sometimes it does not answer at all.

If the order service waits for the email:

1. The customer waits. The payment page gives no answer for a few seconds.
2. An email error breaks the order. The email server is down, so placing the order fails too.
3. On sale day, everything gets slow. A thousand orders in one minute means a thousand emails in that same minute.

The solution: the order service only puts a message in a queue and answers fast. The email service takes the messages from the queue at its own speed. RabbitMQ is a Message Broker that holds these queues.

The main idea: the sender does not know the queue

In RabbitMQ, the sender does not put the message directly in a queue. It gives the message to an Exchange. There are three main ideas:

1. Queue. It holds messages until a consumer takes them. Each service usually has its own queue.
2. Exchange. It takes the message and decides which queues it goes to. It does not hold any messages itself.
3. Binding. A rule that says “this queue wants messages with this key”.

Each message has a Routing Key. For example, the order placed message has the key order.placed, and the order cancelled message has the key order.cancelled.

Why is this separation useful?

1. The sender stays simple. The order service only says “an order was placed”. It does not know how many services are listening.
2. A new service without changing the sender. The report service only creates a new queue and a new Binding.
3. Each queue is independent. If the stock service is off for an hour, its messages stay in its own queue. Emails continue with no problem.

Exchange types

• The direct type. The message key must be exactly the same as the Binding key. For when each kind of work has its own queue.
• The fanout type. It does not look at the key. The message goes to all connected queues. For news that everyone must hear.
• The topic type. The key is compared with a pattern. The star sign means exactly one word, and the # sign means zero or more words. For business events, it is usually the best choice.
• The headers type. Instead of the key, it compares the message Headers. It is used less often.

Several consumers on one queue

On sale day, one copy of the email service is not enough. So we connect three pods of the email service to the same queue.

1. Each message reaches only one of them. RabbitMQ spreads the messages between the consumers of a queue. We call this pattern Competing Consumers.
2. The number of consumers has no structural limit. Unlike Kafka, you do not need to match the number of pods to the number of Partitions.
3. But order is not guaranteed. When several consumers work at the same time, or a message goes back to the queue, the processing order is different from the sending order.

Confirming a message (Ack)

When is a message removed from the queue? When the consumer says “I am done”. We call this an Ack.

• The Ack case. The work succeeded. The message is removed.
• The Nack case without going back to the queue. The work is not possible. If the queue has a Dead Letter Exchange, the message goes to the error queue.
• No answer. If the consumer dies, the message goes back to the queue and reaches another consumer.

Important result: a message may arrive twice. For example, the email was sent, but the service died before the Ack. So the consumer must be Idempotent, exactly like with Kafka.

Limit on messages at the same time (Prefetch)

If you set no limit, RabbitMQ gives any number of unconfirmed messages to one consumer. One pod gets hundreds of messages and the other pods sit idle. With the Prefetch setting, you say “give each consumer at most this many unconfirmed messages”.

Code

New versions of the RabbitMQ.Client library have async methods. First we declare the queue and the Exchange, and send the order placed message:

using RabbitMQ.Client;
using System.Text;
using System.Text.Json;

var factory = new ConnectionFactory { HostName = "localhost" };
using var connection = await factory.CreateConnectionAsync();
using var channel = await connection.CreateChannelAsync();

await channel.ExchangeDeclareAsync("shop.orders", ExchangeType.Topic, durable: true);

await channel.QueueDeclareAsync(
    queue: "email-queue", durable: true, exclusive: false, autoDelete: false,
    arguments: new Dictionary<string, object?>
    {
        ["x-queue-type"] = "quorum",                     // replicated, safe queue
        ["x-dead-letter-exchange"] = "shop.orders.dlx"  // rejected messages go here
    });

await channel.QueueBindAsync("email-queue", "shop.orders", routingKey: "order.*");

var body = Encoding.UTF8.GetBytes(JsonSerializer.Serialize(new OrderPlaced("o-3", 250_000)));
var props = new BasicProperties { Persistent = true, MessageId = "o-3" };

await channel.BasicPublishAsync("shop.orders", "order.placed",
    mandatory: true, basicProperties: props, body: body);

public sealed record OrderPlaced(string OrderId, decimal Total);

Now the email service. We send the Ack ourselves, after the work:

await channel.BasicQosAsync(prefetchSize: 0, prefetchCount: 10, global: false);

var consumer = new AsyncEventingBasicConsumer(channel);
consumer.ReceivedAsync += async (_, ea) =>
{
    try
    {
        var order = JsonSerializer.Deserialize<OrderPlaced>(ea.Body.Span)!;
        await emailSender.SendReceiptOnceAsync(order); // idempotent by OrderId
        await channel.BasicAckAsync(ea.DeliveryTag, multiple: false);
    }
    catch (Exception)
    {
        // Do not requeue forever: send it to the dead letter exchange.
        await channel.BasicNackAsync(ea.DeliveryTag, multiple: false, requeue: false);
    }
};

await channel.BasicConsumeAsync("email-queue", autoAck: false, consumer: consumer);

Do not lose messages

To make sure a message is not lost when the RabbitMQ server restarts, you need several things together:

1. A durable queue. The queue definition stays after a restart.
2. A Persistent message. The message itself is written to disk.
3. A Quorum queue. The message is copied to several servers. If one server dies, the queue is not lost.
4. Publisher Confirms. RabbitMQ tells the sender that it really got the message. Without it, the sender does not know if the message arrived.
5. Manual confirm by the consumer. The autoAck setting must be off, and the Ack must be sent after the work.

Common mistakes

RabbitMQ or Kafka?