5

我正在尝试编写一个服务器/客户端脚本，其中包含一个用于发泄任务的服务器以及多个执行它的工作人员。问题是我的呼吸机有很多任务，它会在心跳中填满内存。我尝试在绑定之前设置 HWM，但没有成功。只要工人连接，它就会继续发送消息，完全无视设置的 HWM。我还有一个水槽，用来记录已完成的任务。

服务器.py

import zmq

def ventilate():
    context = zmq.Context()

    # Socket to send messages on
    sender = context.socket(zmq.PUSH)
    sender.setsockopt(zmq.SNDHWM, 30) #Big messages, so I don't want to keep too many in queue
    sender.bind("tcp://*:5557")


    # Socket with direct access to the sink: used to syncronize start of batch
    sink = context.socket(zmq.PUSH)
    sink.connect("tcp://localhost:5558")

    print "Sending tasks to workers…"

    # The first message is "0" and signals start of batch
    sink.send('0')
    print "Sent starting signal"

    while True:
        sender.send("Message")



if __name__=="__main__":
    ventilate()

工人.py

import zmq
from multiprocessing import Process

def work():
    context = zmq.Context()

    # Socket to receive messages on
    receiver = context.socket(zmq.PULL)
    receiver.connect("tcp://localhost:5557")

    # Socket to send messages to
    sender = context.socket(zmq.PUSH)
    sender.connect("tcp://localhost:5558")

    # Process t asks forever
    while True:
        msg = receiver.recv_msg()
        print "Doing sth with msg %s"%(msg)     
        sender.send("Message %s done"%(msg))

if __name__ == "__main__":
    for worker in range(10):        
        Process(target=work).start()

接收器.py

import zmq

def sink():
    context = zmq.Context()

    # Socket to receive messages on
    receiver = context.socket(zmq.PULL)
    receiver.bind("tcp://*:5558")

    # Wait for start of batch
    s = receiver.recv()
    print "Received start signal"
    while True:
        msg = receiver.recv_msg()
        print msg


if __name__=="__main__":
    sink()
4

2 回答 2

5

好的，我玩了一下，我不认为问题出在 PUSH HWM 上，而是你不能为 PULL 设置 HWM。如果您查看此文档，您会看到它说 N/A for action on HWM。

PULL 套接字似乎每个都接收数百条消息（我确实尝试设置一个 HWM，以防万一它在 PULL 套接字上做了任何事情。它没有。）。我通过更改呼吸机以发送带有递增整数的消息来证明这一点，并将池中的每个工作人员更改为在两次调用之间等待 2 秒recv()。工作人员打印出他们正在处理具有很大不同整数的消息。例如，一个工作人员将处理消息 10，而下一个工作人员正在处理消息 400。随着时间的推移，您会看到正在处理消息 10 的工作人员现在正在处理消息 11、12、13 等。其他是处理401、402等。

这向我表明 ZMQ_PULL 套接字正在某处缓冲消息。因此，虽然 ZMQ_PUSH 套接字确实有一个 HWM，但 PULL 套接字正在快速请求消息，尽管它们实际上并没有通过调用recv(). 因此，如果连接了 PULL 套接字，则会有效地忽略 PUSH HWM。据我所见，您无法控制 PULL 套接字缓冲区的长度（我希望 RCVHWM 套接字选项可以控制它，但似乎没有）。

这种行为当然引出了一个问题，即 ZMQ_PULL HWM 选项的意义何在，只有在您还可以控制接收套接字 HWM 时才有意义。

在这一点上，我会开始询问0MQ 人员，您是否遗漏了一些明显的东西，或者这是否被认为是一个错误。

对不起，我无法提供更多帮助！

于 2014-01-17T00:46:14.490 回答
1

ZeroMQ 在套接字的发送端和接收端都有缓冲区，因此您需要在代码中的 PUSH 和 PULL 套接字上设置高水位线（实际上是在 a bind()or之前connect()）。

在 Python 绑定中，现在可以很方便地通过它一次性socket.hwm = 1设置两者。ZMQ_SNDHWMZMQ_RCVHWM

于 2014-08-01T23:21:52.380 回答