Модель подписки на события, основанная на событиях

Question

Я работаю над требованием, согласно которому процесс (скажем, производитель) должен отправлять односторонние сообщения переменному числу процессов (скажем, потребителям).

Модель публикации-подписки казалась хорошей для этогопотому что потребители будут подписываться на сообщения от производителя.Я пытался использовать ZeroMQ для достижения этой цели.

Однако у меня есть несколько проблем с этим:

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

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

3. Поскольку потребители опрашивают, а сообщения не удаляются из очереди, потребители видят повторяющиеся сообщения, пока не поступит новое сообщение. Я хочу, чтобы потребитель получал уведомление только один раз за новоесообщение.

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

Кто-нибудь может предложить хорошую альтернативу?

Answer 1

Я предлагаю перейти на модель Push-Pull с брокером между производителем и потребителем.

1. Брокер должен быть уведомлен о любом новом сообщении.
2. Потребители будут прислушиваться к уведомлениям брокеров (ведите таблицу для отслеживания успеха / неудачи. Поэтому при повторных попытках избежать дублирования)
3. Как только # 2 завершен, потребитель может получить данные от источника (источника) и отправить подтверждение брокеру для успеха / неудачи

Надеюсь, это поможет

Answer 2

Промежуточное программное обеспечение DDS (Служба распространения данных) поддерживает именно то, что вы пытаетесь достичь, и гораздо проще.

Непосредственно отвечая на ваши вопросы:

1. DDS поддерживает механизм прослушивания, вашим подписчикам не нужно непрерывно опрашивать.

2. DDS имеет хорошие настройки QoS для предотвращения заполнения очередей издателя. Вы можете использовать History QoS, чтобы сказать «сохранить только последние 10 выборок в очереди», или вы можете использовать Lifespan QoS, чтобы сказать «сохранить только образцы, опубликованные за последние 10 секунд».

3. Опять же, вы можете использовать механизм прослушивания DDS, и вы будете уведомлены только один раз для каждого нового семпла. Нет необходимости в опросе.

В настоящее время существует две реализации с открытым исходным кодом.

• OpenSplice: https://github.com/OpenSplice/opensplice
• OpenDDS http://www.opendds.org/

Answer 3

Попробуйте использовать JMS-провайдера или AMQP-провайдера. Вот некоторые из вещей, которые вы ищете с Темами:

1. Push-уведомление для подписчиков.

2. Атрибут времени жизни сообщений, который позволяет удалять сообщения или помещать их в очередь недоставленных сообщений, если они не используются в TTL.

3. Однократное уведомление - в зависимости от вашей конфигурации.

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

С точки зрения того, какой поставщик предоставляет для использования. RabbitMQ популярен для AMQP. Для JMS существует любое количество проприетарных продуктов или реализаций с открытым исходным кодом.

Answer 4

Вам нужно более одного производителя? Если нет, вы можете использовать PUSH / PULL вместо PUB / SUB.

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

Как вы описали, имея потребителей в качестве конечной точки SUB, вы можете в конечном итоге доставлять одно и то же сообщение более чем одному потребителю (при условии, что это будет проблемой в вашей модели), если два или более потребителей подписаны на один и тот же "префикс" .

Предполагается, что "префикс" - это строка, которую вы передаете sock.setsockopt(ZMQ_SUBSCRIBE, "prefix", ...);