具有qt事件循环的pyzmq

Question

我在我的qt应用程序中使用pyzmq。

我在第一环节中的mailinglist中找到了一些过去的解决方案。这是我的链接代码。

import zmq
from PyQt5.QtCore import QSocketNotifier
from PyQt5.QtWidgets import QApplication, QWidget


class ChatApp(QWidget):
    def __init__(self):
        super(ChatApp, self).__init__()

        self._zmq_context = zmq.Context()
        self._zmq_sock = self._zmq_context.socket(zmq.SUB)
        self._zmq_sock.connect("tcp://localhost:5556")
        self._zmq_sock.setsockopt(zmq.SUBSCRIBE, b"bm_chat")

        self.read_noti = QSocketNotifier(self._zmq_sock.getsockopt(zmq.FD),
                                             QSocketNotifier.Read,
                                             self)
        self.read_noti.activated.connect(self.on_read_msg)

    def on_read_msg(self, _):
        self.read_noti.setEnabled(False)
        flags = self._zmq_sock.getsockopt(zmq.EVENTS)

        if flags & zmq.POLLIN:
            msg = self._zmq_sock.recv_multipart()
            topic = msg[0]
            data = msg[1]
            print(topic, data)
        elif flags & zmq.POLLOUT:
            print("[Socket] zmq.POLLOUT")
        elif flags & zmq.POLLERR:
            print("[Socket] zmq.POLLERR")
        else:
            print("[Socket] FAILURE")
        self.read_noti.setEnabled(True)

if __name__ == '__main__':
    app = QApplication([])
    win = ChatApp()
    win.show()
    app.exec_()

但是，正如预期后的消息火一次，再也不会发生。这是我的口信

[Socket] FAILURE
b'bm_trade' b'hello0'

因此，在启用qt通知之后，我使用read self._zmq_sock.getsockopt(zmq.EVENTS)搜索了其他解决方案。所以我修改了最后一行的代码

def on_read_msg(self, _):
    self.read_noti.setEnabled(False)
    flags = self._zmq_sock.getsockopt(zmq.EVENTS)

    if flags & zmq.POLLIN:
        msg = self._zmq_sock.recv_multipart()
        topic = msg[0]
        data = msg[1]
        print(topic, data)
    elif flags & zmq.POLLOUT:
        print("[Socket] zmq.POLLOUT")
    elif flags & zmq.POLLERR:
        print("[Socket] zmq.POLLERR")
    else:
        print("[Socket] FAILURE")
    self.read_noti.setEnabled(True)
    self._zmq_sock.getsockopt(zmq.EVENTS) // Here is fixed      

它工作得很好，直到数据速率很低。这是我的酒吧服务器代码。

from time import sleep

import zmq

context = zmq.Context()
socket = context.socket(zmq.PUB)
socket.bind("tcp://*:5557")

def ms(millisec):
    return millisec / 1000

if __name__ == '__main__':
    count = 0
    while True:
        socket.send_multipart(
            [b'bm_trade', bytes(('hello' + str(count)).encode('utf-8'))])
        count += 1

        sleep(ms(10))

这个技巧只适用于事件时差超过10毫秒的情况。如果您将睡眠(ms())更改为10以下，客户端也会被解雇一次，而且永远不会触发。

有人会说没关系，但在我的应用程序中，延迟非常低，所有消息都应该在gui中可见。

为什么会发生这个问题，如何解决呢？

Answer 1

当有数据时，您只需继续阅读，如我所示：

def on_read_msg(self):
    self.read_noti.setEnabled(False)

    if self._zmq_sock.getsockopt(zmq.EVENTS) & zmq.POLLIN:
        while self._zmq_sock.getsockopt(zmq.EVENTS) & zmq.POLLIN:
            topic, data = self._zmq_sock.recv_multipart()
            print(topic, data)
    elif self._zmq_sock.getsockopt(zmq.EVENTS) & zmq.POLLOUT:
        print("[Socket] zmq.POLLOUT")
    elif self._zmq_sock.getsockopt(zmq.EVENTS) & zmq.POLLERR:
        print("[Socket] zmq.POLLERR")

    self.read_noti.setEnabled(True)