ZeroMQ 订阅者在单元测试中不接收任何数据。为什么？
A ZeroMQ Subscriber does not receive any data in a unittest. Why?
我不明白我的数据最终去了哪里。
我已经编写了一个测试来确保我的 Publisher class 成功发送数据，并且如果我绑定到它就可以接收到该数据。
class 本身继承自 Thread，并公开了一个 publish() 方法，我可以调用该方法来传递要通过 Queue() 广播给订阅者的数据。
但是，在我的测试中，数据从未到达。我确保使用相同的
端口，我想不出这里还有什么问题。
我是 ZeroMQ 新手，但我之前已经设法让 PubSub 模式正常工作。
测试代码：
# Import Built-ins
import time
import json
import queue
from queue import Queue
from threading import Thread
# Import Third-Party
import zmq
def test_publisher_sends_data(self):
port = 667
name, topic, data = 'TestNode', 'testing', ['this', 'is', 'data']
encoded_name = json.dumps(name).encode('utf-8')
encoded_topic = json.dumps(topic).encode('utf-8')
encoded_data = json.dumps(data).encode('utf-8')
expected_result = (encoded_name, encoded_topic, encoded_data)
publisher = Publisher(port)
print("starting publisher")
publisher.start()
q = Queue()
def recv(q):
ctx = zmq.Context()
zmq_sock = ctx.socket(zmq.SUB)
print("Connecting to publisher")
zmq_sock.connect('tcp://127.0.0.1:%s' % port)
while True:
print("waiting for data..")
q.put(zmq_sock.recv_multipart())
print("data received!")
t = Thread(target=recv, args=(q,))
t.start()
print("sending data via publisher")
for i in range(5):
self.assertTrue(publisher.publish(name, topic, data))
time.sleep(0.1)
print("checking q for received data..")
try:
result = q.get(block=False)
except queue.Empty:
self.fail("Queue was empty, no data received!")
self.assertEqual(expected_result, result)
PublisherClass
# Import Built-Ins
import logging
import json
from queue import Queue
from threading import Thread, Event
# Import Third-Party
import zmq
class Publisher(Thread):
"""Publisher Class which allows publishing data to subscribers.
The publishing is realized with ZMQ Publisher sockets, and supports publishing
to multiple subscribers.
The run() method continuosly checks for data on the internal q, which is fed
by the publish() method.
"""
def __init__(self, port, *args, **kwargs):
"""Initialize Instance.
:param port:
"""
self.port = port
self._running = Event()
self.sock = None
self.q = Queue()
super(Publisher, self).__init__(*args, **kwargs)
def publish(self, node_name, topic, data):
"""Publish the given data to all current subscribers.
All parameters must be json-serializable objects
:param data:
:return:
"""
message_parts = [json.dumps(param).encode('utf-8')
for param in (node_name, topic, data)]
if self.sock:
self.q.put(message_parts)
return True
else:
return False
def join(self, timeout=None):
self._running.clear()
try:
self.sock.close()
except Exception:
pass
super(Publisher, self).join(timeout)
def run(self):
self._running.set()
ctx = zmq.Context()
self.sock = ctx.socket(zmq.PUB)
self.sock.bind("tcp://*:%s" % self.port)
while self._running.is_set():
if not self.q.empty():
msg_parts = self.q.get(block=False)
print("Sending data:", msg_parts)
self.sock.send_multipart(msg_parts)
else:
continue
ctx.destroy()
self.sock = None
添加.setsockopt( zmq.SUBSCRIBE, someNonZeroLengthSTRING )
作为
已记录的默认 SUB-套接字实例未订阅任何内容
（顺其自然）
如果任何传入的消息无法匹配任何字符串，SUB 端被订阅，本地 .recv() 自然不会收到这样的消息。
鉴于您的代码没有显式订阅，因此不存在满足主题过滤器处理条件的消息，Q.E.D。
最好的下一步：
作为另一个问题 -- "Late-joiner" 麻烦 -- 接下来可能会发生，如果单元测试设计是快速的，我可能会建议您进一步（不仅是 ZeroMQ ）分布式系统设计的最佳下一步, 是花时间阅读 Pieter HINTJENS 的精彩著作 "Code Connected, Volume 1"。任何认真研究异构分布式系统信号/消息传递的人都会喜欢他分享技术和非技术观点和意见。
我不明白我的数据最终去了哪里。
我已经编写了一个测试来确保我的 Publisher class 成功发送数据，并且如果我绑定到它就可以接收到该数据。
class 本身继承自 Thread，并公开了一个 publish() 方法，我可以调用该方法来传递要通过 Queue() 广播给订阅者的数据。
但是，在我的测试中，数据从未到达。我确保使用相同的 端口，我想不出这里还有什么问题。
我是 ZeroMQ 新手，但我之前已经设法让 PubSub 模式正常工作。
测试代码：
# Import Built-ins
import time
import json
import queue
from queue import Queue
from threading import Thread
# Import Third-Party
import zmq
def test_publisher_sends_data(self):
port = 667
name, topic, data = 'TestNode', 'testing', ['this', 'is', 'data']
encoded_name = json.dumps(name).encode('utf-8')
encoded_topic = json.dumps(topic).encode('utf-8')
encoded_data = json.dumps(data).encode('utf-8')
expected_result = (encoded_name, encoded_topic, encoded_data)
publisher = Publisher(port)
print("starting publisher")
publisher.start()
q = Queue()
def recv(q):
ctx = zmq.Context()
zmq_sock = ctx.socket(zmq.SUB)
print("Connecting to publisher")
zmq_sock.connect('tcp://127.0.0.1:%s' % port)
while True:
print("waiting for data..")
q.put(zmq_sock.recv_multipart())
print("data received!")
t = Thread(target=recv, args=(q,))
t.start()
print("sending data via publisher")
for i in range(5):
self.assertTrue(publisher.publish(name, topic, data))
time.sleep(0.1)
print("checking q for received data..")
try:
result = q.get(block=False)
except queue.Empty:
self.fail("Queue was empty, no data received!")
self.assertEqual(expected_result, result)
PublisherClass
# Import Built-Ins
import logging
import json
from queue import Queue
from threading import Thread, Event
# Import Third-Party
import zmq
class Publisher(Thread):
"""Publisher Class which allows publishing data to subscribers.
The publishing is realized with ZMQ Publisher sockets, and supports publishing
to multiple subscribers.
The run() method continuosly checks for data on the internal q, which is fed
by the publish() method.
"""
def __init__(self, port, *args, **kwargs):
"""Initialize Instance.
:param port:
"""
self.port = port
self._running = Event()
self.sock = None
self.q = Queue()
super(Publisher, self).__init__(*args, **kwargs)
def publish(self, node_name, topic, data):
"""Publish the given data to all current subscribers.
All parameters must be json-serializable objects
:param data:
:return:
"""
message_parts = [json.dumps(param).encode('utf-8')
for param in (node_name, topic, data)]
if self.sock:
self.q.put(message_parts)
return True
else:
return False
def join(self, timeout=None):
self._running.clear()
try:
self.sock.close()
except Exception:
pass
super(Publisher, self).join(timeout)
def run(self):
self._running.set()
ctx = zmq.Context()
self.sock = ctx.socket(zmq.PUB)
self.sock.bind("tcp://*:%s" % self.port)
while self._running.is_set():
if not self.q.empty():
msg_parts = self.q.get(block=False)
print("Sending data:", msg_parts)
self.sock.send_multipart(msg_parts)
else:
continue
ctx.destroy()
self.sock = None
添加.setsockopt( zmq.SUBSCRIBE, someNonZeroLengthSTRING )
作为
已记录的默认 SUB-套接字实例未订阅任何内容
（顺其自然）
如果任何传入的消息无法匹配任何字符串，SUB 端被订阅，本地 .recv() 自然不会收到这样的消息。
鉴于您的代码没有显式订阅，因此不存在满足主题过滤器处理条件的消息，Q.E.D。
最好的下一步：
作为另一个问题 -- "Late-joiner" 麻烦 -- 接下来可能会发生，如果单元测试设计是快速的，我可能会建议您进一步（不仅是 ZeroMQ ）分布式系统设计的最佳下一步, 是花时间阅读 Pieter HINTJENS 的精彩著作 "Code Connected, Volume 1"。任何认真研究异构分布式系统信号/消息传递的人都会喜欢他分享技术和非技术观点和意见。