ZeroMQ PULL 获取队列中的所有消息

Question

我有大量（~100）PUSH“客户端”定期（~1 / s）向PULL“服务器”发送消息。服务器以每秒约 2 批的速度分批处理消息。批量大小不是固定的——理想情况下，我想在每次迭代时处理队列中的所有消息。我获取队列中所有消息的代码如下：

buffer = []
while True:
  try:
    buffer.append(socket.recv_pyobj(zmq.NOBLOCK))
  except zmq.ZMQError as e:
    break
print(len(buffer))

这似乎可行，但缓冲区长度在两次迭代之间变化很大。通常，我希望每次阅读约 50 条消息。但是，我经常看到许多迭代少于 10 条消息，然后是一个非常大的消息（约 1000 条）。我怀疑我在缓冲或类似的事情上做错了，因为这不应该发生。

Answer 1

接收方的高度动态变化？

我的怀疑是 PULL 方面的公平队列策略，硬连线在 .Context() 低级数据泵内，它服务于（如下）所有.connect()-ed 对等方的强制工作周期。

除非您能够以某种方式平衡馈送 PUSH-ers 发送各自消息的延迟，否则公平排队接收周期将完全按照您报告的方式，显示那些没有PUSH“及时”他们的下一条消息的人。