Модель подписки на публикацию, управляемую событиями

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

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

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

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

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

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

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

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


design-patterns publish-subscribe zeromq data-distribution-service
person user1259642    schedule 26.03.2012    source источник
comment
Мое единственное предложение - обернуть опрос pub / sub в специальный поток слушателя, который отправляет уведомление в основной цикл приложения (или что-то еще), а также игнорирует повторяющиеся сообщения и как можно быстрее выгружает почтовый ящик.   -  person Aaron Watters    schedule 26.03.2012
comment
Я рекомендую вам ознакомиться со стандартом промежуточного программного обеспечения DDS от OMG. Уже существует хорошая реализация с открытым исходным кодом, она называется OpenDDS (см. opendds.org). Он имеет богатый набор настроек качества обслуживания, которые позволяют контролировать его поведение. Предстоящая версия OpenDDS 3.1 будет очень хорошей.   -  person Johnny Willemsen    schedule 27.03.2012
comment
Меня смущают ваши 3 вопроса о ZMQ pub / sub. Все это похоже на то, что вы действительно не исследовали, что делает ZMQ? 1) Всегда существует какая-то форма опроса, независимо от того, делает ли ваш клиент это явно, или вы прикрепляете его к какому-либо обработчику, а библиотека делает это под капотом 2) ZMQ имеет высокую отметку, когда он начнет отбрасывать сообщения после определенного момента, если они не опустошены из буфера 3) сообщения доставляются каждому подписчику только один раз. Я что-то упускаю?   -  person jdi    schedule 28.03.2012

Ответы (4)


arrow_upward
1
arrow_downward

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

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

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

person Madhusudanan    schedule 28.05.2012

arrow_upward
0
arrow_downward

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

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

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

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

person Dalton Barreto    schedule 28.03.2012
comment
Я думаю, что даже если бы ему понадобилось несколько продюсеров, он все равно мог бы использовать этот подход с устройством посередине. Несколько производителей будут подключаться к устройству, а несколько потребителей будут потреблять его. - person jdi; 28.03.2012

arrow_upward
0
arrow_downward

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

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

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

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

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

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

person scaganoff    schedule 10.04.2012

arrow_upward
0
arrow_downward

Промежуточное ПО DDS (Data Distribution Service) поддерживает именно то, что вы пытаетесь достичь, и намного проще.

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

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

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

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

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

person Ertan D.    schedule 31.05.2012