python – 试图了解zeromq高水印行为

我一直在玩pyzmq并使用HWM进行简单的负载平衡,我不太了解我所看到的行为.

我已经建立了一个简单的多线程测试,DEALER客户端通过ROUTER到DEALER模式连接到两个工作人员. HWM设置为1.其中一个工作人员非常快,另一个工作人员非常慢,而且所有客户端都向服务器发送了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!"

示例输出：

...
0 81
0 82
0 83
0 84
0 85
0 86
0 87
0 88
0 89
0 90
0 91
0 92
0 93
0 94
0 95
0 96
0 97
0 98
0 99
1 1
1 3
1 5

解决方法:

显然,ZeroMQ会从send()调用中异步发送消息.也就是说,当send()返回时,消息尚未发送或添加到内部队列.如果发送速度足够快,则下次调用send时,消息仍未添加到队列中,因此尚未到达水印.您可以在将某些消息放入队列之前添加数十或数百条消息,达到水印,然后阻止发送行为.

换句话说,尝试在send()之后休息几分之一秒,看看会发生什么,它应该给消息添加到队列足够的时间,所以到下次发送时,它能够看到已达到水印.

标签：python,multithreading,zeromq,pyzmq
来源： https://codeday.me/bug/20190629/1323357.html