如何异步接收 ZeroMQ (pyzmq) JSON 消息？答案

【问题标题】：How can I receive a ZeroMQ (pyzmq) JSON message asynchronously?如何异步接收 ZeroMQ (pyzmq) JSON 消息？
【发布时间】：2021-09-24 03:18:50
【问题描述】：

现在我的代码中有以下结构。请求直接解析为 JSON。

while True:
   if self.rest_sock.poll(timeout=0):
      request = self.rest_sock.recv_json()
      ...

我想用函数的异步调用来替换循环（以减少 CPU 时间，如下所述：https://stackoverflow.com/a/21937311/10555800）。这是通过使用on_recv() 将函数注册为事件处理程序来完成的。但是 JSON 消息没有被解析。我假设我可以自己解析它，例如解释这里https://stackoverflow.com/a/34242555/10555800。但我想知道为什么没有与 socket.recv_json() 等效的东西来异步接收像 on_recv_json() 这样的 json 消息。

编辑（回答@bazza 的问题）：

与此同时，我的程序还有其他事情要做。这确实导致了一些其他问题，我已经通过将超时设置为 None 解决了这个问题。
我想我也可以按照您的建议删除投票，因为与套接字连接相关的所有内容都在其自己的线程中运行，因此不应阻塞整个程序

【问题讨论】：

所以，我想我们又回到了您已经掌握的代码 sn-p 上。如果没有就绪消息，则超时为零的轮询立即返回，并且脚本继续执行其他操作。如果这不是太长并且本质上是循环的（while 循环），您很快就会回到投票中。如果这表明一条消息现在已经准备好了，那么这个线程中唯一可能发生的事情就是 JSON 解析，因为 ZMQ 自己的线程已经处理了跨网络的消息读取。
或者，完全摆脱轮询，并调用套接字接收传递适当的标志以实现非阻塞。

标签： python asynchronous zeromq pyzmq

【解决方案1】：

代码中还有什么内容？

只有在没有可用的入站消息时，如果程序还有其他事情要做，则只能使用超时 0 进行轮询。当轮询返回表明没有套接字准备好时，让程序执行其他操作
如果您需要对准备好读取的几个套接字中的任何一个做出反应，并且在此期间无事可做，您可以轮询多个套接字（请参阅here），超时时间不为零。
如果您只有一个套接字，并且您只是在等待消息到达，然后再执行任何其他操作，只需调用recv()，它将阻塞直到出现消息。这种情况下不需要轮询。

了解字典中“poll”的定义是“检查状态”的意思是很有帮助的。

【讨论】：

我的问题可以用这个建议解决，但它没有回答是否有类似on_recv JSON 方法的问题。
@Tobias，我认为on_recv_json() 函数没有任何意义。 on_recv() 注册一个回调，但它本身并不读取套接字。至于回调的作用，这与on_recv() 无关，因此JSON 编码的消息是无关紧要的。回调将调用socket.recv_json()。此外，消息是 JSON 也与 ZMQ 无关 - recv_json() 将只是一个方便的函数，它使用 recv() 接收消息，然后在消息上调用 JSON 解析器。
@Tobias，另外，我在答案中的建议是您可能根本不需要 on_recv() 。这完全取决于您希望您的应用程序是Pro-actor 还是Reactor。如果您打算使用 zmq 轮询，那就是 Reactor，并且您不注册回调，您只需在轮询器说消息到达时调用 socket.recv()。如果你打算使用 on_recv()，那就是Proactor，你也不会使用 zmq 轮询器。将Proactor 和Reactor 混合在一个应用程序中会带来很多困难！
好的，非常感谢！我认为您是正确的，在我的情况下没有必要使用on_recv。