试图了解 zeromq 高水位标记行为答案

【问题标题】：trying to understand zeromq high water mark behaviour试图了解 zeromq 高水位标记行为
【发布时间】：2014-02-16 22:02:33
【问题描述】：

我一直在使用 pyzmq 和使用 HWM 进行简单的负载平衡，但我不太了解我所看到的行为。

我设置了一个简单的多线程测试，一个 DEALER 客户端通过 ROUTER 到 DEALER 模式连接到两个工作人员。 HWM 设置为 1。其中一个 worker 非常快，另一个非常慢，客户端所做的只是向服务器发送 100 条垃圾邮件。这通常似乎有效，并且较快的工作人员处理的消息比速度较慢的工作人员多得多。

但是，即使我将慢速工作人员设置得如此缓慢，这样快速工作人员应该能够在慢速工作人员完成甚至一条消息之前处理 99 条消息，但慢速工作人员似乎仍然收到至少 2 或 3 条消息.

是高水位线行为不准确还是我遗漏了什么？

服务器代码如下：

import re, sys, time, string, zmq, threading, signal


def worker_routine(worker_url, worker_id, context=None):
    # socket to talk to dispatcher
    context = context or zmq.Context.instance()
    socket = context.socket(zmq.REP)
    socket.set_hwm(1)
    socket.connect(worker_url)

    print "worker ", worker_id, " ready ..."
    while True:
        x = socket.recv()

        if worker_id==1:
            time.sleep(3)

        print worker_id, x
        sys.stdout.flush()

        socket.send(b'world')


context = zmq.Context().instance()
# socket facing clients
frontend = context.socket(zmq.ROUTER)
frontend.bind("tcp://*:5559")
# socket facing services
backend  = context.socket(zmq.DEALER)
url_worker = "inproc://workers"
backend.set_hwm(1)
backend.bind(url_worker)

# launch pool of worker threads
for i in range(2):
    thread = threading.Thread(target=worker_routine, args=(url_worker,i,))
    thread.start()
    time.sleep(0.1)

try:
    zmq.device(zmq.QUEUE, frontend, backend)
except:
    print "terminating!"

# we never get here
frontend.close()
backend.close()
context.term()

客户端代码如下：

import zmq, random, string, time, threading, signal

#  prepare our context and sockets
context = zmq.Context()
socket = context.socket(zmq.DEALER)
socket.connect("tcp://localhost:5559")

inputs = [''.join(random.choice(string.ascii_lowercase) for x in range(12)) for y in range(100)]

for x in xrange(100):
    socket.send_multipart([b'', str(x)])

print "finished!"

示例输出：

【问题讨论】：

标签： python multithreading zeromq pyzmq

【解决方案1】：

显然 ZeroMQ 从您的 send() 调用中异步发送消息。也就是说，当 send() 返回时，消息还没有发送出去，或者还没有加入到内部队列中。如果发送速度够快，下次调用 send 时，消息仍然没有被加入队列，因此还没有达到水印。您可以添加数十或数百条消息，然后一些消息进入队列，达到水印，阻止发送行为开始。

换句话说，尝试在 send() 之后休眠几分之一秒，看看会发生什么，它应该有足够的时间将消息添加到队列中，所以到下一次发送时，它是可以看到水印已经到达。

【讨论】：