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

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

Answer 1

嗯，好的，找到答案了。

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

在这种情况下，我使用 ROUTER/DEALER 架构意味着当我执行 send_json 时，从 DEALER 套接字发送的 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 向另一个方向发送消息时，您必须将其作为多部分发送，指定它的目的地是哪个客户端，例如:

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