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

这是我的代码，去除了多余的内容:

协调器

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()

worker.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中的错误吗？我使用错了吗？

最佳答案

嗯，好的，找到了答案。

ØMQ接口(interface)中有一个令人讨厌的不对称性，因此您必须了解所使用的套接字的类型。

在这种情况下，我使用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”部分来进一步简化代码。因此， worker 只是这样做:

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向另一个方向发送消息时，您必须将其作为多部分发送，并指定要发送给哪个客户端，例如:

协调器

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方式来处理广播消息，而不是像我在上面那样循环发送给每个客户端。我希望文档中对每种可用的套接字类型以及如何使用它们都有清晰的描述。