проблема при реализации неблокирующего механизма публикации с помощью zeromq

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

Пытаясь реализовать это с помощью паттерна PUB-SUB, я заметил, что если получатель не запущен, отправитель зависает. Например, в следующем исходном коде отправителя:

import zmq

context = zmq.Context()
sender = context.socket(zmq.PUB)
sender.connect("tcp://localhost:5555")

sender.send("Sending to nobody", NOBLOCK)

print "Msg sent"

когда получатель не работает, сообщение «Сообщение отправлено» никогда не печатается, а отправитель остается в «sender.send («Отправка никому», NOBLOCK)» навсегда. Кроме того, я попытался проверить, работает ли приемник или не получает возврат функции connect, но в обоих случаях всегда "Нет".

Я использую Python 2.6.5 и zeromq 2.1.

Кто-нибудь знает, что это происходит или альтернативное решение? (Я пытался использовать PULL-PUSH и REQ-REP, но результаты были схожими)

Заранее большое спасибо


python nonblocking zeromq
person corto    schedule 07.11.2011    source источник
comment
Этот код на самом деле не работает (символ NOBLOCK не определен). Это действительно помогает размещать код, который люди могут копировать и вставлять.   -  person larsks    schedule 07.11.2011

Ответы (2)


arrow_upward
3
arrow_downward

После исправления образца:

import zmq

context = zmq.Context()
sender = context.socket(zmq.PUB)
sender.connect("tcp://localhost:5555")

sender.send("Sending to nobody", zmq.NOBLOCK)

print "Msg sent"

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

Вы можете изменить это поведение, установив параметр LINGER в вашем сокете:

sender = context.socket(zmq.PUB)
sender.setsockopt(zmq.LINGER, 100)

Здесь это значение представляет собой время задержки в миллисекундах. См. справочную страницу для zmq_setsockopt для получения дополнительной информации. Практическим результатом этого является то, что ZMQ будет ждать только задержка миллисекунд перед закрытием сокета.

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

person larsks    schedule 07.11.2011
comment
Почему sender.close() и context.term() не возвращают код ошибки, когда получатель не работает, а zmq.LINGER равно 100 или больше? Или как я могу теперь, когда отправка сообщения не удалась? - person Ib33X; 09.01.2012

arrow_upward
3
arrow_downward

Вы можете легко изменить: sender.connect("tcp://localhost:5555") на sender.bind("tcp://localhost:5555") и наоборот на приемнике.

Это дает вам возможность подключить приемник в любом месте. Отправитель будет работать независимо от того, есть получатели или нет.

person Oleg Fedoseev    schedule 07.11.2011