zeromq: как предотвратить бесконечное ожидание?

Если вы используете zeromq >= 3.0, вы можете установить параметр сокета RCVTIMEO:

client_receiver.RCVTIMEO = 1000 # in milliseconds

Но в целом можно использовать поллеры:

poller = zmq.Poller()
poller.register(client_receiver, zmq.POLLIN) # POLLIN for recv, POLLOUT for send

И poller.poll() берет тайм-аут:

evts = poller.poll(1000) # wait *up to* one second for a message to arrive.

evts будет пустым списком, если нечего получать.

Вы можете опросить с помощью zmq.POLLOUT, чтобы проверить, будет ли отправка успешной.

Или, чтобы обработать случай с узлом, который мог выйти из строя, a:

worker.send(msg, zmq.NOBLOCK)

может быть достаточно, что всегда будет возвращаться немедленно, вызывая ZMQError(zmq.EAGAIN), если отправка не может быть завершена.

Это был быстрый хак, который я сделал после того, как сослался на ответ пользователя938949 и http://taotetek.wordpress.com/2011/02/02/python-multiprocessing-with-zeromq/. Если у вас получится лучше, опубликуйте свой ответ, я порекомендую ваш ответ.

Тем, кто ищет долговременные решения по надежности, см. http://zguide.zeromq.org/page:all#toc64

Zeromq версии 3.0 (бета-банкомат) поддерживает время ожидания в ZMQ_RCVTIMEO и ZMQ_SNDTIMEO. http://api.zeromq.org/3-0:zmq-setsockopt

Сервер

Zmq.NOBLOCK гарантирует, что если клиент не существует, send() не блокируется.

import time
import zmq
context = zmq.Context()

ventilator_send = context.socket(zmq.PUSH)
ventilator_send.bind("tcp://127.0.0.1:5557")

i=0

while True:
    i=i+1
    time.sleep(0.5)
    print ">>sending message ",i
    try:
        ventilator_send.send(repr(i),zmq.NOBLOCK)
        print "  succeed"
    except:
        print "  failed"

Клиент

Объект опроса может прослушивать многие принимающие сокеты (см. «Многопроцессорная обработка Python с ZeroMQ», ссылка на которую приведена выше. Я связал его только с work_receiver. В бесконечном цикле клиент опрашивает с интервалом в 1000 мс. Объект socks возвращает пустое значение, если за это время не было получено ни одного сообщения.

import time
import zmq
context = zmq.Context()

work_receiver = context.socket(zmq.PULL)
work_receiver.connect("tcp://127.0.0.1:5557")

poller = zmq.Poller()
poller.register(work_receiver, zmq.POLLIN)

# Loop and accept messages from both channels, acting accordingly
while True:
    socks = dict(poller.poll(1000))
    if socks:
        if socks.get(work_receiver) == zmq.POLLIN:
            print "got message ",work_receiver.recv(zmq.NOBLOCK)
    else:
        print "error: message timeout"

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

Причина в том, что сокет ожидает любого однорангового узла, чтобы исходящие сообщения попадали в очередь. Чтобы немедленно закрыть сокет и сбросить исходящие сообщения из буфера, используйте ZMQ_LINGER и установите его в 0..

Если вы ждете только один сокет, а не создаете Poller, вы можете сделать это:

if work_receiver.poll(1000, zmq.POLLIN):
    print "got message ",work_receiver.recv(zmq.NOBLOCK)
else:
    print "error: message timeout"

Вы можете использовать это, если ваш тайм-аут меняется в зависимости от ситуации, вместо установки work_receiver.RCVTIMEO.