pyzmq recv_json 无法解码 send_json 发送的消息

Question

这是我的代码，去掉了无关的东西：

协调器.py

context = zmq.Context()
socket = context.socket(zmq.ROUTER)
port = socket.bind_to_random_port(ZMQ_ADDRESS)

poller = zmq.Poller()
poller.register(socket, zmq.POLLIN)

while True:
    event = poller.poll(1)
    if not event:
        continue
    process_id, val = socket.recv_json()

工人.py

context = zmq.Context()
socket = context.socket(zmq.DEALER)
socket.connect('%s:%s' % (ZMQ_ADDRESS, kwargs['zmq_port']))

socket.send_json(
    (os.getpid(), True)
)

当我运行它时会发生什么：

    process_id, val = socket.recv_json()
  File "/Users/anentropic/.virtualenvs/myproj/lib/python2.7/site-packages/zmq/sugar/socket.py", line 380, in recv_json
    return jsonapi.loads(msg)
  File "/Users/anentropic/.virtualenvs/myproj/lib/python2.7/site-packages/zmq/utils/jsonapi.py", line 71, in loads
    return jsonmod.loads(s, **kwargs)
  File "/Users/anentropic/.virtualenvs/myproj/lib/python2.7/site-packages/simplejson/__init__.py", line 451, in loads
    return _default_decoder.decode(s)
  File "/Users/anentropic/.virtualenvs/myproj/lib/python2.7/site-packages/simplejson/decoder.py", line 406, in decode
    obj, end = self.raw_decode(s)
  File "/Users/anentropic/.virtualenvs/myproj/lib/python2.7/site-packages/simplejson/decoder.py", line 426, in raw_decode
    raise JSONDecodeError("No JSON object could be decoded", s, idx)
JSONDecodeError: No JSON object could be decoded: line 1 column 0 (char 0)

如果我深入研究 ipdb：

> /Users/anentropic/.virtualenvs/myproj/lib/python2.7/site-packages/zmq/sugar/socket.py(380)recv_json()
    379             msg = self.recv(flags)
--> 380             return jsonapi.loads(msg)
    381

ipdb> p msg
'\x00\x9f\xd9\x06\xa2'

嗯，这看起来不像 JSON……这是 pyzmq 中的错误吗？我用错了吗？

Answer 1

嗯，好的，找到答案了。

ØMQ 接口中存在令人讨厌的不对称性，因此您必须了解您正在使用的套接字类型。

在这种情况下，我使用 ROUTER/DEALER 架构意味着从 DEALER 套接字发送的 JSON 消息，当我这样做时send_json，被包裹在多部分消息信封中。第一部分是客户端 ID（我猜这是'\x00\x9f\xd9\x06\xa2'我上面得到的），第二部分是我们感兴趣的 JSON 字符串。

所以在我的 coordinator.py 的最后一行，我需要这样做：

id_, msg = socket.recv_multipart()
process_id, val = json.loads(msg)

恕我直言，这是 ØMQ/pyzmq 的糟糕设计，库应该将其抽象出来，并拥有刚刚send和recv可以工作的方法。

我从这个问题中得到了线索How can I use send_json with pyzmq PUB SUB所以看起来 PUB/SUB 架构有同样的问题，毫无疑问其他人也有。

这在文档中有所描述，但不是很清楚
http://zguide.zeromq.org/page:all#The-Asynchronous-Client-Server-Pattern

更新

事实上，我发现在我的情况下，我可以通过直接使用消息信封的“客户端 ID”部分来进一步简化代码。所以工人只是做：

context = zmq.Context()
socket = context.socket(zmq.DEALER)
socket.identity = str(os.getpid())  # or I could omit this and use ØMQ client id
socket.connect('%s:%s' % (ZMQ_ADDRESS, kwargs['zmq_port']))

socket.send_json(True)

还值得注意的是，当您想从 ROUTER 向另一个方向发送消息时，您必须将其作为多部分发送，指定它的目的地是哪个客户端，例如：

协调器.py

context = zmq.Context()
socket = context.socket(zmq.ROUTER)
port = socket.bind_to_random_port(ZMQ_ADDRESS)

poller = zmq.Poller()
poller.register(socket, zmq.POLLIN)

pids = set()
while True:
    event = poller.poll(1)
    if not event:
        continue
    process_id, val = socket.recv_json()
    pids.add(process_id)

    # need some code in here to decide when to stop listening
    # and break the loop

for pid in pids:
    socket.send_multipart([pid, 'a string message'])
    # ^ do your own json encoding if required

我想可能有一些 ØMQ 方式来做广播消息，而不是像上面那样循环发送到每个客户端。我希望文档对每种可用的套接字类型以及如何使用它们有一个清晰的描述。