Сервис заказов публикует события, а получателей уже несколько: письма, склад, аналитика. У каждого свои интересы: письмам нужны только оплаты, склад работает с одним регионом, аналитике нужно всё. Кто решает, какое сообщение в какую очередь попадёт? Это работа exchange. В этом уроке разберём его типы — direct, fanout, topic и headers, научимся строить подписки, а потом посмотрим, что делать, когда обработка сообщения не удаётся: для этого в RabbitMQ есть dead-letter очереди, TTL и повторы с задержкой.
Как exchange принимает решение
Вспомним цепочку: producer → exchange → binding → очередь → consumer. Exchange ничего не хранит: он смотрит на ключ маршрутизации (routing key) сообщения и на привязки (binding) и копирует сообщение в подходящие очереди. Привязка — это пара «очередь + ключ привязки» (binding key). Тип exchange определяет, как сравниваются эти два ключа:
| Тип | Правило | Пример |
|---|---|---|
direct | ключи совпадают точно | уровень лога: error |
fanout | ключ игнорируется, копия уходит всем | событие для всех сервисов |
topic | ключ сопоставляется с шаблоном | order.*.paid |
headers | сравниваются заголовки, а не ключ | format=pdf |
direct: точное совпадение
Пример — логи по уровням. Ошибки (error) нужны очереди оповещений, а info и warning — консоли:
import amqp from 'amqplib';
const connection = await amqp.connect('amqp://localhost');
const channel = await connection.createChannel();
await channel.assertExchange('logs', 'direct', { durable: true });
await channel.assertQueue('alerts', { durable: true });
await channel.assertQueue('console', { durable: true });
await channel.bindQueue('alerts', 'logs', 'error');
await channel.bindQueue('console', 'logs', 'info');
await channel.bindQueue('console', 'logs', 'warning');
const send = (level, text) =>
channel.publish('logs', level, Buffer.from(text));
send('error', 'Payment service is down');
send('info', 'Order 7 created');
send('trace', 'nobody listens to this level');Сообщение с ключом error попадёт в alerts, info — в console. А trace, для которого нет привязки, молча пропадёт: exchange не знает, куда его отправить, и отбрасывает. Чтобы замечать такие потери, у publish есть флаг mandatory, а у exchange — настройка alternate exchange.
💡 Если привязать к одному ключу несколько очередей, каждая получит свою копию:
directумеет и это.
fanout: копия всем
fanout игнорирует ключ и копирует каждое сообщение во все привязанные очереди. Это самый простой pub/sub из первого урока модуля. Подписчик объявляет свою очередь и привязывает её к exchange:
