python - ZeroMQ PUB 套接字在连接时缓冲我所有的外出数据

Question

例如，我注意到 zeromq PUB 套接字在连接时会缓冲所有传出数据

import zmq
import time
context = zmq.Context()

# create a PUB socket
pub = context.socket (zmq.PUB)
pub.connect("tcp://127.0.0.1:5566")
# push some message before connected
# they should be dropped
for i in range(5):
    pub.send('a message should not be dropped')

time.sleep(1)

# create a SUB socket
sub = context.socket (zmq.SUB)
sub.bind("tcp://127.0.0.1:5566")
sub.setsockopt(zmq.SUBSCRIBE, "")

time.sleep(1)

# this is the only message we should see in SUB
pub.send('hi')

while True:
    print sub.recv()

sub 在这些消息之后绑定(bind)，它们应该被丢弃，因为如果没有人连接到它，PUB 应该丢弃消息。但它不会丢弃消息，而是缓冲所有消息。

a message should not be dropped
a message should not be dropped
a message should not be dropped
a message should not be dropped
a message should not be dropped
hi

如您所见，那些“不应丢弃的消息”由套接字缓冲，一旦连接，它就会将它们刷新到 SUB 套接字。如果我在 PUB 套接字处绑定(bind)，并在 SUB 套接字处连接，那么它可以正常工作。

import zmq
import time
context = zmq.Context()

# create a PUB socket
pub = context.socket (zmq.PUB)
pub.bind("tcp://127.0.0.1:5566")
# push some message before connected
# they should be dropped
for i in range(5):
    pub.send('a message should not be dropped')

time.sleep(1)

# create a SUB socket
sub = context.socket (zmq.SUB)
sub.connect("tcp://127.0.0.1:5566")
sub.setsockopt(zmq.SUBSCRIBE, "")

time.sleep(1)

# this is the only message we should see in SUB
pub.send('hi')

while True:
    print repr(sub.recv())

而且你只能看到输出

'hi'

这种奇怪的行为会导致问题，它会在一个连接套接字上缓冲所有数据，我有两个服务器，服务器 A 将数据发布到服务器 B

Server A -- publish --> Server B

如果服务器 B 上线，它工作正常。但是，如果我启动了服务器 A 而没有启动服务器 B 呢？

结果，服务器A上连接的PUB套接字保留了所有这些数据，内存使用率越来越高。

问题来了，这种行为是错误还是特性？如果是功能，我在哪里可以找到提到此行为的文档？以及如何停止连接 PUB 套接字缓冲所有数据？

谢谢。

Answer 1

套接字是阻塞还是丢弃消息取决于套接字类型，如 ZMQ::Socket documentation 中所述(下面的重点是我的):

ZMQ::HWM: Retrieve high water mark

The ZMQ::HWM option shall retrieve the high water mark for the specified socket. The high water mark is a hard limit on the maximum number of outstanding messages 0MQ shall queue in memory for any single peer that the specified socket is communicating with.

If this limit has been reached the socket shall enter an exceptional state and depending on the socket type, 0MQ shall take appropriate action such as blocking or dropping sent messages. Refer to the individual socket descriptions in ZMQ::Socket for details on the exact action taken for each socket type.

The default ZMQ::HWM value of zero means “no limit”.

您可以通过查看 ZMQ::HWM option action 的套接字类型的文档来查看它是否会阻塞或丢弃，这将是 Block 或 放下。

ZMQ::PUB 的操作是Drop，所以如果它没有被丢弃，你应该检查 HWM(高水位线)值并注意警告 < strong>默认的 ZMQ::HWM 值为零意味着“无限制”，意味着它不会进入异常状态，直到系统内存耗尽(此时我不知道它的行为如何) .