我需要为 ZMQ 上的每个订阅创建新的套接字吗？

Question

服务器循环遍历对象列表，这些对象上的数据实时变化。服务器每毫秒发布这些对象的所有新数据。 即['Carrot', 'Banana', 'Mango', 'Eggplant']

客户端可以通过名称订阅特定对象。 self.sub_socket.setsockopt_string(zmq.SUBSCRIBE, 'Carrot')
在线程上，客户端也会实时轮询这些数据

while True:
    sockets = dict(self.poller.poll(poll_timeout))
    if self.sub_socket in sockets and sockets[self.sub_socket] == zmq.POLLIN:
        msg = self.sub_socket.recv_string(zmq.DONTWAIT)

        // do something with the msg...

问题是当我订阅多个对象时，比如说Carrot, Eggplant & Banana。我只是 收到来自Carrot 的更改，有时是Banana，在Eggplant 上很少见。我认为这是因为从服务器的循环顺序来看，比如当客户端轮询时，接收 Carrot，处理数据，然后再次轮询，但服务器已经完成了通过列表发布并再次发布 Carrot 然后客户端因此，民意调查只收到了胡萝卜。

所以我想为每个订阅创建单独的套接字？这是一个解决方案吗？我是 ZMQ 的新手，希望能得到一些帮助，谢谢！

Answer 1

Q：“这是一个解决方案吗？” _{...为每个订阅创建单独的套接字？}

不会。除非出于某些我不知道的其他原因。

---

_{如果您从未与ZeroMQ 合作过，
可以在这里先看看"ZeroMQ: Principles in less than Five Seconds"
在深入了解更多细节之前}

---

虽然 ZeroMQ 消息传递基础架构为每个消息传递提供零保修，但这并不意味着消息在发送后会在任何地方蒸发或丢失。它只是说，期望每个交付的产品都有零保修，如果有人需要，可以添加这种想要的保修机制开销，如果其他人可以在没有它们的情况下工作，则无需支付费用。失去 1-in-1.000.000？ 1 合 1.000.000.000？这取决于许多因素，但丢失消息并不是distributed-computing 系统的常见或随机状态（并且有一些内部原因，其细节超出了本文的范围）。

---

还有疑问？

做一个测试。

设计一个简单的测试——PUB——发送一个均匀分布的琐碎消息

SAMPLEs  = int( 1E6 )
aMsgSIZE = 2048
TOPICs   = [ r'Carrot', r'Banana', r'Mango', r'Eggplant', r'' ]
MASK     = "{0:}" + aMsgSIZE * "_"

for i in range( SAMPLEs ):
    PUB.send( MASK.format( TOPICs[np.random.randint( len( TOPICs ) - 1 ) ) )
    time.sleep( 1E-3 )

使用此测试，您将收到相同数量的均匀分布样本，每个订阅的TOPICs（如果全部订阅）。

增加aMsgSIZE 可能（默认为Context()- 和Socket()-instances）创建一些消息以“丢失”，但同样，这应该是均匀分布的。如果没有，深入挖掘会有些麻烦。

统一未从SAMPLEs 发送的消息量将表明需要调整Context() 和Socket()-instances 的参数以提供足够的资源来安全地排队该数据量-流动。然而，为单个订阅的 Topic 字符串提供更多 Socket()-s 并不能解决这个资源管理瓶颈（如果存在）。

不要犹豫，发布测试结果，如果主题的均匀分布组合是否偏斜，以及最后没有收到多少部分。

添加平台详细信息 + ZeroMQ 版本，一如既往：o)