Как безопасно "повесить трубку" ZeroMQ?

Я начал использовать ZeroMQ на этой неделе, и при использовании шаблона Request-Response я не уверен, как заставить работника безопасно "повесить трубку" и закрыть его сокет, не отбрасывая сообщение и заставляя клиента, который отправил это сообщение, никогда не получайте ответа. Представьте себе работника, написанного на Python, который выглядит примерно так:

import zmq
c = zmq.Context()
s = c.socket(zmq.REP)
s.connect('tcp://127.0.0.1:9999')
while i in range(8):
    s.recv()
    s.send('reply')
s.close()

Я экспериментировал и обнаружил, что клиент в 127.0.0.1:9999 типа сокета zmq.REQ, который делает запрос с честной очередью, может иметь несчастье, если алгоритм честной очереди выберет вышеупомянутого работника сразу после работник выполнил последний send(), но перед тем, как запустить следующий метод close(). В этом случае кажется, что запрос получен и буферизирован стекю ØMQ в рабочем процессе и что запрос затем теряется, когда close() выкидывает все, что связано с сокетом.

Как рабочий может отсоединиться "безопасно" - есть ли способ сигнализировать "Я больше не хочу сообщений", затем (а) перебирать любые окончательные сообщения, которые были получены во время передачи сигнала, (b) сгенерировать их ответы, а затем (c) выполнить close() с гарантией того, что сообщения не будут выброшены?

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

Изменить:. В ответ на хороший вопрос исправлено описание, чтобы количество ожидающих сообщений было множественным, так как при ответах было много подключений.

Ответ 1

Кажется, вы думаете, что пытаетесь избежать "простого" состояния гонки, например, в

... = zmq_recv(fd);
do_something();
zmq_send(fd, answer);
/* Let hope a new request does not arrive just now, please close it quickly! */
zmq_close(fd);

но я думаю, что проблема заключается в том, что честная очередность (round-robin) усложняет ситуацию: у вас может даже быть несколько запросов в очереди для вашего рабочего. Отправитель не будет ждать, пока ваш работник станет свободным, прежде чем отправлять новый запрос, если его очередь будет получена, поэтому в то время, когда вы вызываете zmq_send, другие запросы могут уже ждать.

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

Конечно, это означает использование XREP/XREQ, но я думаю, что это того стоит.

Изменить: Я написал некоторый код, реализующий в другом направлении, чтобы объяснить, что я имею в виду.

Ответ 2

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

Некоторые люди делают это с помощью PUB/SUB для контроля, чтобы каждый работник публиковал acks, и мастер подписывается на них.

Вы должны помнить, что с ZeroMQ существует 0 очереди сообщений. Вовсе нет! Просто сообщения буферизуются либо отправителем, либо получателем в зависимости от настроек, таких как метка высокой воды, и типа сокета. Если вам действительно нужны очереди сообщений, вам нужно написать приложение-брокер, чтобы справиться с этим, или просто переключиться на AMQP, где все общение осуществляется через стороннего брокера.

Ответ 3

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

Ответ 4

Существует конфликт интересов между отправкой запросов как можно быстрее для работников и получением надежности в случае сработанных сбоев или вымираний. Существует целый раздел руководства ZeroMQ, в котором объясняются разные ответы на этот вопрос надежности. Прочтите, что это поможет.

tl; dr работники могут/будут разбиваться, а клиентам нужна функция повторной отправки. Руководство предоставляет код многократного использования для этого на многих языках.

Ответ 5

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

Ответ 6

Попробуйте спящий режим перед закрытием вызова. Это исправлено в 2.1, но не в 2.0.