Типы обменников в rabbitmq и хранение сообщения при неактивном получателе?

Question

[deadsquirrel93]

В оф. доке rabbitmq написано, что есть несколько типов обменников(exchange), один из них fanout - распределяет сообщения во все очереди, игнорирует routing_key.
Модель задачи: есть n сообщений, есть три получателя(consumer), каждое сообщение должно попасть к каждому получателю.
Используется либа https://github.com/php-amqplib/php-amqplib

Что делаю:
Отправитель(producer): декларирую обменник типа fanout, делаю basic_publish всех сообщений
Получатель(consumer): декларирую обменник типа fanout декларирую очередь с рандомным названием(из доки list($queue_name, ,) = $channel->queue_declare("", false, false, true, false);), делаю basic_consume всех сообщений

Проблема:
Если получатели уже запущены, то все происходит верно, все получатели получили по каждому сообщению, но если получатели не запущены, а сообщения отправлены, то при запуске получателей ничего не происходит, сообщения просто потерялись.
Если декларировать очередь в отправителе, то сообщения не теряются, но потом не привязываясь в получателе к этой очереди я не смогу взять сообщения и весь смысл fanout теряется. Игрался с разными флагами(durable, delivery_mode) и ничего не помогло.

Вопрос: как сохранять сообщения, не привязываясь к очереди в отправителе, чтобы при запуске получателей они стали получать эти сообщения? Если способ только декларировать очередь в отправителе, то тогда какой смысл fanout обменника?

Answer 1

[Сергей]

Если Вы хотите на пустом RabbitMQ создать обменник, кинуть в него сообщения и позже подключить очереди к этому обменнику в надежде получить ранее отправленные сообщения, то это у Вас не получиться. Обменник не хранит сообщения, а разбрасывает входящие сразу по мере их получения, если в данный момент у него нет очередей для отправки - сообщение просто отбрасывается. Вам нужно стартовать потребителей до того, как отправлять сообщения. Если хотите, чтобы в момент, когда потребитель отвалился, сообщения накапливались, то надо сделать предсказуемое/известное для потребителя имя очереди (по необходимости сделать её durable).
Если надо, чтобы при старте RabbitMQ что-то создавало очереди и обменники, то через менеджмент плагин такое можно устроить (говорят, в последних версиях даже можно это сделать без менеджмент плагина).

Comments

[deadsquirrel93]

Т.е. если стартовать потребителя, забить обменник сообщениями и стопнуть потребителя, то сообщения еще будут накапливаться в обменнике, до тех пор, пока заново не запустится потребитель?

Получается в любом случае нужно знать имя очереди потребителя, очередь отправителя декларировать необязательно, запустить потребителя перед отправителем с известным именем очереди, поставить отправку сообщений отправителя и можно стопнуть потребителя и сообщения будут накапливаться в его очереди?

[deadsquirrel93]

Спрошу еще проще(чтобы самому было понятнее):
1) Публикуются сообщения с заказами(1, 2 ... n) без декларирования очереди
2) Тип обменника fanout
3) В каждом потребителе создаю по одной очереди с известными именами(internal, external, mobile)
4) Запускаю эти три разных файла(воркеры?) и жду сообщений
Предположим пришло 4 заказа, все 4 потребителя их получили, потом потребитель external отвалился, но пришло еще 10 заказов, при включении потребителя external он начнет получать заказы с 5 по 14, в то время как остальные потребители уже получили заказы?

[Sergei Iamskoi]

deadsquirrel93, если вы декларировали очереди, то верно. Когда он подключиться, то дополучит сообщения. Имейте ввиду что у сообщений есть срок жизни ttl, а также если погасить сервер то они пропадут. Вроде кролик поддерживает возможность дампить недоставленные сообщения на диск, что бы в таком случае они не пропадали, но ценой скорости работы.

fanout наверное это больше для кластера: приходит сообщение в один эксчейнж, а из него распределяется по доступным все кластерам. А далее уже обрабатывается согласно routing_key. Сам fanout routing_key не обрабатывает, но он его прокидывает дальше, и далее он конечно может быть обработан.

[Сергей]

deadsquirrel93, да, если очередь существует (и привязана к обменнику) в момент публикации сообщения, то сообщение в ней останется до того как потребитель запросит его (через subscription или basic_get, при условии что нет ограничивающих время политик).
Исправлю немного ответ выше, fanout это самый просто тип обменника и он может применяться вне зависимоти кластер это или одна нода, это зависит от задачи, если нужно "продублировать" сообщение во все очереди - он идеален для этого. Так же никто не мешает биндить один обменник на другой.

Answer 2

[index0h]

как сохранять сообщения, не привязываясь к очереди в отправителе, чтобы при запуске получателей они стали получать эти сообщения?

Создать эксчейндж, создать очереди, забиндить очереди на эксчейндж.

Если способ только декларировать очередь в отправителе, то тогда какой смысл fanout обменника?

Смысл в том, что распределение идет по очередям, а не консьюмерам. Если нет очередей - распределять некуда.

fanout вам не нужен