Блог

Требование

Система претерпевает некоторое изменение состояния, и об этом должны знать несколько других частей системы (назовем их наблюдателями), чтобы они могли выполнять некоторые действия на основе текущего состояния, действия наблюдателей важны, если некоторые наблюдатели не подключены к Сети (в настоящее время не прослушиваются из-за некоторыхпроблема, но скоро вернется), сообщение не следует отбрасывать, пока все наблюдатели не получат сообщение.

Пытаясь выполнить это с помощью модели pub / sub, вот мои выводы (пожалуйста, исправьте, если это понимание неверно) —

Издатель создает событие по определенной теме, и несколько подписчиков могут использовать одно и то же сообщение. Эта модель либо не предоставляет гарантии доставки (в redis), либо доставка гарантируется один раз (с очередями обмена сообщениями), т.е. когда один из потребителей подтверждает сообщение, сообщение отбрасывается (rabbitmq).

Пример

В БД создается новый объект профиля пользователя

Теперь,

1. Служба фоновой проверки должна знать это, чтобы запустить процесс проверки.
2. Служба подписок должна знать это, чтобы добавить подписки по умолчанию для пользователя.

Теперь обе задачи важны, не связаны и могут выполняться параллельно.

Теперь в модели очереди, если служба подписки по какой-либо причине отключена, процесс проверки BG подтверждает сообщение, сообщение будет удалено из очереди, или, если оно запущено и забыто, как и большинство pub / sub, доставка в любом случае не гарантируется для обеих служб.

Еще один момент — обе задачи не связаны и не должны запускаться одна за другой.

Короче говоря, мне нужно убедиться, что все потребители получают одно и то же сообщение, и они должны иметь возможность подтверждать их по отдельности, сообщение должно быть удалено только после того, как все потребители признали это, ни один из вышеперечисленных подходов этого не делает.

Чего-нибудь мне здесь не хватает? Как мне подойти к этой проблеме?

Этот сценарий явно поддерживается моделью RabbitMQ, которая отделяет «обмены» от «очередей»:

• Издатель всегда отправляет сообщение на «exchange», который является просто адресом маршрутизации без сохранения состояния; ему не нужно знать, в какие очереди должно попасть сообщение
• Потребитель всегда считывает сообщения из «очереди», которая содержит свою собственную копию сообщений, независимо от того, откуда они возникли
• Несколько потребителей могут подписаться на одну и ту же очередь, и каждое сообщение будет доставлено только одному потребителю
• Важно отметить, что обмен может направлять одно и то же сообщение в несколько очередей, и каждая получит копию сообщения

Здесь важно понять, что, хотя мы говорим о потребителях, «подписывающихся» на очередь, частью «подписки» в настройке «pub-sub» на самом деле является маршрутизация из exchange в очередь.

Таким образом, система RabbitMQ pub-sub может выглядеть следующим образом:

1. В БД создается новый объект профиля пользователя
2. Это событие публикуется как сообщение в разделе «события» exchange с ключом маршрутизации «entity.profile.created»
3. Обмен направляет копии сообщения в несколько очередей:
   • К этому обмену была привязана очередь «verification_service» для получения копии всех сообщений, соответствующих «entity.profile.#»
   • К этому обмену была привязана очередь «subscription_setup_service» для получения копии всех сообщений, соответствующих «entity.profile.created»
4. Сценарии-потребители ничего не знают об этой маршрутизации, они просто знают, что сообщения будут появляться в очереди для событий, которые имеют отношение к ним:
   • Служба проверки получает копию сообщения в очереди «verification_service», обрабатывает и подтверждает ее
   • Служба настройки подписки получает копию сообщения в очереди «subscription_setup_service», обрабатывает и подтверждает его
5. Если есть несколько сценариев-пользователей, просматривающих одну и ту же очередь, они будут делиться сообщениями в этой очереди между собой, но при этом полностью независимы от любой другой очереди.

Вот скриншот из этого интерактивного инструмента визуализации, который показывает этот сценарий:

Как вы упомянули, это не то, что вы можете контролировать с помощью структуры данных Redis Pub / Sub.

Но вы можете сделать это легко с помощью потоков Redis.

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

Вы можете посмотреть на эти примеры приложений, которые предоставляют (на Java) пример о:

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

Ссылки:

• Начало работы с потоками Redis и Java
• Потоки Redis в действии ( проект, который показывает, как использовать ADD / ACK / PENDING / CLAIM и создавать потоковое приложение с защитой от ошибок с помощью потоков Redis и SpringData)