PySide6中QUdpSocket不触发readyRead 等效Qt C++代码运行正常

核心问题原因 #

• 未校验bind操作返回值，端口被占用、地址无效等bind失败场景未被感知
• 绑定的地址限制过严：绑定127.0.0.1仅能接收本机环回口的UDP数据包，非本机发过来的数据包会被直接丢弃
• Windows平台默认绑定标识存在限制：未显式开启地址复用的情况下，端口被其他进程占用时会绑定失败
• 若代码中额外调用了connectToHost进入ConnectedState：UDP无连接特性下，调用connectToHost后receiveDatagram接口会失效，只能用read()系列接口读取数据，且仅能读取到目标地址/端口的数据包

修正后可运行代码 #

import sys

from PySide6.QtCore import QObject, Signal
from PySide6.QtGui import QIntValidator
from PySide6.QtNetwork import QUdpSocket, QHostAddress
from PySide6.QtWidgets import QApplication, QDialog, QGridLayout, QLabel, QLineEdit

class UdpClient(QDialog):
    def __init__(self, parent= None):
        super().__init__(parent)

        self._block_size = 0
        self._current_fortune = ''

        host_label = QLabel("&Server name:")
        port_label = QLabel("S&erver port:")

        self._host_line_edit = QLineEdit('Localhost')
        self._port_line_edit = QLineEdit()
        self._port_line_edit.setValidator(QIntValidator(1, 65535, self))

        host_label.setBuddy(self._host_line_edit)
        port_label.setBuddy(self._port_line_edit)

        self._status_label = QLabel("Data: ")

        self.hostname = "0.0.0.0" # 改为监听所有网卡
        self.portnum = 12000

        self._udp_socket_manager = UdpConectionManager(self.hostname, self.portnum)
        self._udp_socket_manager.ProcessTheDataGram.connect(self.process_Data)

        main_layout = QGridLayout(self)
        main_layout.addWidget(self._status_label, 2, 0, 1, 2)

        self.setWindowTitle("Fortune Client")
        self._port_line_edit.setFocus()

    def process_Data(self, data):
        self._status_label.setText(data)


class UdpConectionManager(QObject):
    ProcessTheDataGram = Signal(str)
    
    def __init__(self, hostname, portnum):
        super().__init__()
        self.udp_socket_ptr = QUdpSocket(self)
        print("Making socket connection to : ", portnum)
        # 显式指定绑定标识，增加绑定成功校验
        bind_result = self.udp_socket_ptr.bind(
            QHostAddress(hostname), 
            int(portnum),
            QUdpSocket.ShareAddress | QUdpSocket.ReuseAddressHint
        )
        if not bind_result:
            print(f"绑定失败，错误信息：{self.udp_socket_ptr.errorString()}")
            return
        self.udp_socket_ptr.readyRead.connect(self.readPendingDataGrams)
        print(f"绑定成功，正在监听端口{portnum}")

    def readPendingDataGrams(self):
        while self.udp_socket_ptr.hasPendingDatagrams():
            datagram = self.udp_socket_ptr.receiveDatagram()
            # 字节转utf8字符串，避免显示b''前缀
            data_str = datagram.data().decode('utf-8', errors='replace')
            self.ProcessTheDataGram.emit(data_str)


if __name__ == '__main__':
    app = QApplication(sys.argv)
    client = UdpClient()
    client.show()
    sys.exit(client.exec())

额外注意事项 #

• 如果仅需要和固定的服务端通信，调用connectToHost后，将receiveDatagram替换为readAll()读取数据即可
• Windows系统下运行需要放行防火墙对当前Python进程的网络访问权限，否则UDP数据包会被防火墙拦截
• 测试时可以用网络调试助手向本机12000端口发送UDP数据包验证接收效果

内容的提问来源于stack exchange，提问作者bg1420