当前位置：首页 > news >正文

PyQt学习系列06-网络编程与通信协议

news 2025/5/24 7:59:24

PyQt学习系列笔记（Python Qt框架）

第六课：PyQt的网络编程与通信协议

一、网络编程基础

1.1 网络通信的核心概念

网络编程是实现客户端-服务器通信的基础，PyQt通过QtNetwork模块提供完整的网络功能支持。

核心概念：

TCP（传输控制协议）：面向连接、可靠、基于字节流的协议，适用于需要数据完整性的场景（如文件传输、远程登录）。
UDP（用户数据报协议）：无连接、不可靠、基于数据报的协议，适用于实时性要求高的场景（如音视频传输、游戏）。
HTTP/HTTPS：基于TCP的应用层协议，用于Web通信（如API请求、网页加载）。
WebSocket：全双工通信协议，支持客户端与服务器实时交互（如在线聊天、实时数据更新）。

二、PyQt网络编程的核心模块

2.1 QtNetwork模块

PyQt的QtNetwork模块提供了以下关键类：

类名	功能
`QTcpSocket`	实现TCP客户端和服务器端通信。
`QUdpSocket`	实现UDP通信（广播、多播）。
`QHttp`	提供HTTP协议支持（已弃用，推荐使用`QNetworkAccessManager`）。
`QNetworkAccessManager`	管理HTTP/HTTPS请求，支持异步通信。
`QNetworkRequest`	定义网络请求的参数（如URL、头部）。
`QNetworkReply`	表示网络请求的响应结果。

三、TCP通信实现

3.1 TCP服务器端实现

使用QTcpServer监听端口并处理客户端连接。

示例：TCP服务器

from PyQt5.QtNetwork import QTcpServer, QTcpSocket, QHostAddress
from PyQt5.QtCore import QObject, pyqtSignal, pyqtSlotclass TcpServer(QObject):client_connected = pyqtSignal(str)  # 客户端连接信号def __init__(self, port=8080):super().__init__()self.server = QTcpServer(self)self.server.listen(QHostAddress.Any, port)self.server.newConnection.connect(self.handle_new_connection)@pyqtSlot()def handle_new_connection(self):client_socket = self.server.nextPendingConnection()client_ip = client_socket.peerAddress().toString()self.client_connected.emit(f"新连接: {client_ip}")client_socket.readyRead.connect(lambda: self.read_data(client_socket))client_socket.disconnected.connect(lambda: self.handle_disconnection(client_socket))@pyqtSlot()def read_data(self, socket):data = socket.readAll().data().decode()print(f"收到数据: {data}")socket.write(f"服务器响应: {data}".encode())@pyqtSlot()def handle_disconnection(self, socket):print("客户端断开连接")socket.deleteLater()

3.2 TCP客户端实现

使用QTcpSocket连接服务器并发送数据。

示例：TCP客户端

from PyQt5.QtNetwork import QTcpSocket
from PyQt5.QtCore import QObject, pyqtSignal, pyqtSlotclass TcpClient(QObject):data_received = pyqtSignal(str)  # 数据接收信号def __init__(self, host='127.0.0.1', port=8080):super().__init__()self.socket = QTcpSocket(self)self.socket.connected.connect(self.on_connected)self.socket.readyRead.connect(self.read_data)self.socket.errorOccurred.connect(self.handle_error)self.socket.connectToHost(host, port)@pyqtSlot()def on_connected(self):print("已连接到服务器")self.socket.write("Hello, Server!".encode())@pyqtSlot()def read_data(self):data = self.socket.readAll().data().decode()self.data_received.emit(data)@pyqtSlot(QTcpSocket.SocketError)def handle_error(self, error):print(f"错误: {self.socket.errorString()}")

四、UDP通信实现

4.1 UDP广播/单播

QUdpSocket支持发送和接收UDP数据包，适用于广播和单播场景。

示例：UDP广播

from PyQt5.QtNetwork import QUdpSocket
from PyQt5.QtCore import QObject, pyqtSignalclass UdpBroadcaster(QObject):def __init__(self, port=8888):super().__init__()self.udp_socket = QUdpSocket(self)self.udp_socket.bind(QHostAddress.Any, port)self.udp_socket.readyRead.connect(self.read_data)def send_broadcast(self, message):self.udp_socket.writeDatagram(message.encode(), QHostAddress.Broadcast, 8888)def read_data(self):while self.udp_socket.hasPendingDatagrams():data, host, port = self.udp_socket.readDatagram(self.udp_socket.pendingDatagramSize())print(f"收到消息: {data.decode()} from {host}:{port}")

五、多线程网络通信

5.1 使用`QThread`避免UI卡顿

长时间运行的网络操作应放在子线程中，避免阻塞主线程（UI线程）。

示例：网络请求线程

from PyQt5.QtCore import QThread, pyqtSignal
import requestsclass NetworkThread(QThread):response_received = pyqtSignal(str)  # 响应信号def __init__(self, url):super().__init__()self.url = urldef run(self):try:response = requests.get(self.url)self.response_received.emit(response.text)except Exception as e:self.response_received.emit(f"错误: {e}")

六、HTTP/HTTPS请求

6.1 使用`QNetworkAccessManager`

QNetworkAccessManager支持异步HTTP请求，适合构建现代Web应用。

示例：GET请求

from PyQt5.QtNetwork import QNetworkAccessManager, QNetworkRequest, QNetworkReply
from PyQt5.QtCore import QUrl, QObject, pyqtSignalclass HttpManager(QObject):response_received = pyqtSignal(str)  # 响应信号def __init__(self):super().__init__()self.manager = QNetworkAccessManager(self)self.manager.finished.connect(self.handle_response)def get(self, url):request = QNetworkRequest(QUrl(url))self.manager.get(request)def handle_response(self, reply: QNetworkReply):if reply.error() == QNetworkReply.NoError:data = reply.readAll().data().decode()self.response_received.emit(data)else:self.response_received.emit(f"错误: {reply.errorString()}")reply.deleteLater()

七、WebSocket通信

7.1 实时双向通信

QWebSocket支持全双工通信，适用于在线聊天、实时数据更新等场景。

示例：WebSocket客户端

from PyQt5.QtWebSockets import QWebSocket
from PyQt5.QtCore import QObject, pyqtSignalclass WebSocketClient(QObject):message_received = pyqtSignal(str)  # 消息信号def __init__(self, url):super().__init__()self.socket = QWebSocket()self.socket.connected.connect(self.on_connected)self.socket.textMessageReceived.connect(self.handle_message)self.socket.open(QUrl(url))def on_connected(self):print("已连接到WebSocket服务器")self.socket.sendTextMessage("Hello, WebSocket!")def handle_message(self, message):self.message_received.emit(message)