RabbitMQ: exchanges, queues, routing, ack
↑ 5.4 Очереди и брокеры: Kafka, RabbitMQ · ← Предыдущая · → Следующая
Зачем это на собесе
RabbitMQ — классический брокер для задач и маршрутизации; нужно знать модель exchange → queue.
Объяснение
Продюсер публикует в exchange, тот по правилам маршрутизации (binding) раскладывает сообщения по очередям, потребители читают из очередей.
| Тип exchange | Маршрутизация |
|---|---|
| direct | по точному routing key |
| fanout | во все связанные очереди |
| topic | по шаблону (orders.*.created, #) |
| headers | по заголовкам |
flowchart LR P[Producer] --> E{Exchange} E -- key=orders.created --> Q1queue billing E -- key=orders.* --> Q2queue analytics Q1 --> C1[Consumer] Q2 --> C2[Consumer]
Ключевые механизмы:
- Ack/Nack: потребитель подтверждает обработку; без ack сообщение возвращается в очередь.
- Prefetch (QoS): сколько неподтверждённых сообщений держит потребитель.
- Durable + persistent: очередь и сообщения переживают перезапуск брокера.
- Publisher confirms: брокер подтверждает приём продюсеру.
- TTL, DLX (dead letter exchange), приоритеты, delayed messages.
- Quorum queues: реплицируемые очереди на Raft (рекомендуются вместо classic mirrored).
var factory = new ConnectionFactory { HostName = "rabbit", DispatchConsumersAsync = true };
using var conn = await factory.CreateConnectionAsync();
using var ch = await conn.CreateChannelAsync();
await ch.QueueDeclareAsync("billing", durable: true, exclusive: false, autoDelete: false);
await ch.BasicQosAsync(0, 10, false);
var consumer = new AsyncEventingBasicConsumer(ch);
consumer.ReceivedAsync += async (_, ea) =>
{
await Handle(ea.Body);
await ch.BasicAckAsync(ea.DeliveryTag, false);
};
await ch.BasicConsumeAsync("billing", autoAck: false, consumer);Нюансы и подводные камни
autoAck: true— потеря при сбое обработчика.- Канал не потокобезопасен: один канал на поток/потребителя.
- Без ограничения очередь растёт неограниченно и «съедает» память.
- Порядок гарантируется только при одном потребителе на очередь без requeue.
- Сообщения удаляются после подтверждения — переиграть историю нельзя (в отличие от Kafka).
Практика
- Настройте topic exchange с двумя очередями и разными шаблонами.
- Реализуйте ack/nack и DLX для необработанных сообщений.
- Сравните throughput с разным prefetch.
Вопросы с ответами
Чем exchange отличается от очереди?
Exchange маршрутизирует сообщения, очередь их хранит.
Что произойдёт без ack?
Сообщение считается недоставленным и вернётся в очередь при закрытии канала/таймауте.
Как сделать pub/sub в RabbitMQ?
Fanout/topic exchange и отдельная очередь у каждого подписчика.