ZeroMQ：有人可以解释一下 HWM 是如何工作的吗？

Question

我正在试验 ZeroMQ，我有一个 PUSH 应用程序，它发送约 370,000 条消息，每条消息大小为 10Kb。

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

no_msgs =      0
i       = 370000

fd  = open( "input_text_files/file1.txt", 'r' )
msg = fd.read()

while i >  0:
    print "Sending data of size" , len( msg )
    rc       = socket.send( msg )
    no_msgs += 1
    if rc == None:
        print True
    if rc == -1:
        print "ERROR!!"

    i = i-1

PULL 应用程序刚刚循环接收消息：

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

no_msgs = 1
i       = 1

while True:
    msg      = socket.recv()
    no_msgs += 1
    print "Received msg", i
    print "Received %s" %( len( msg ) )
    i = i + 1

使用此设置，有消息丢失，即PULL 套接字没有收到所有 370,000 条消息。我在PUSH 套接字上添加了socket.hwm = 10，并且收到了所有370,000 条消息。有人可以向我解释这是如何工作的吗？

Answer 1

嗯，HighWaterMarks 是一组微妙的设置：

这些在 ZeroMQ .Context() 实例的“底层”工作

可能是，开始阅读一些有关 ZeroMQ 内部的内容，接下来尝试区分非阻塞和阻塞操作模式。

ZMQ_RCVBUF: Set kernel receive buffer size | default -1 == O/S def'd
ZMQ_RCVHWM: Set high water mark for inbound messages | default 1000
ZMQ_SNDBUF: Set kernel transmit buffer size | default -1 == O/S def'd
ZMQ_SNDHWM: Set high water mark for outbound messages| default 1000

ZMQ_SNDHWM 选项应为指定套接字上的出站消息设置高水位线。高水位线是对未完成消息的最大数量的硬性限制，ØMQ 应在内存中为指定套接字正在与之通信的任何单个对等方排队。零值表示没有限制。

如果已达到此限制，则套接字将进入异常状态，并且根据套接字类型，ØMQ 应采取适当的措施，例如阻塞或丢弃发送的消息。请参阅zmq_socket(3) 中的各个套接字描述，了解对每种套接字类型采取的具体操作的详细信息。

ØMQ 不保证套接字会接受尽可能多的ZMQ_SNDHWM 消息，实际限制可能会降低多达 60-70%，具体取决于套接字上的消息流.

ZMQ_PUSH：

管道节点使用ZMQ_PUSH 类型的套接字将消息发送到下游管道节点。消息被轮询到所有连接的下游节点。此套接字类型未实现zmq_recv() 函数。

当ZMQ_PUSH套接字进入静音状态时由于所有下游节点都达到了高水位线，或者如果根本没有下游节点，然后任何zmq_send(3) 套接字上的操作都将阻塞，直到静音状态结束或至少一个下游节点可用于发送；消息不会被丢弃。

ZMQ_RECVHWM: ^{...PULL边设置}

ZMQ_RCVHWM 选项应为指定套接字上的入站消息设置高水位标记。 高水位线是对指定套接字正在与之通信的任何单个对等方的 ØMQ 应在内存中排队的未完成消息的最大数量的硬限制。零值表示没有限制。
如果已达到此限制，则套接字将进入异常状态，并且根据套接字类型，ØMQ 应采取适当的措施，例如阻塞或丢弃发送的消息。请参阅zmq_socket(3) 中的各个套接字描述，了解对每种套接字类型采取的具体操作的详细信息。