如何使用PyQt6的QTcpSocket实现非阻塞式本地服务器连接与通信？

核心逻辑说明 #

Qt的网络模块本身是事件驱动非阻塞的，无需额外开线程，只要通过QTcpSocket的信号监听连接状态、数据接收等事件，就能在处理网络操作的同时显示加载动画。核心要用到的信号包括：

• connected：连接成功触发
• readyRead：有数据可读取时触发
• errorOccurred：连接/通信出错时触发
• disconnected：连接断开时触发

完整代码实现 #

以下是基于PyQt6的示例，PyQt5只需调整导入路径即可：

from PyQt6.QtWidgets import (
    QMainWindow, QWidget, QVBoxLayout, QLineEdit,
    QPushButton, QLabel, QTextEdit
)
from PyQt6.QtNetwork import QTcpSocket, QAbstractSocket
from PyQt6.QtCore import QByteArray, QJsonDocument, QJsonObject, QMovie
import sys

class ChatWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("聊天室")
        self.resize(400, 500)
        
        # 聊天界面控件
        central_widget = QWidget()
        self.setCentralWidget(central_widget)
        layout = QVBoxLayout(central_widget)
        
        self.chat_display = QTextEdit()
        self.chat_display.setReadOnly(True)
        layout.addWidget(self.chat_display)
        
        self.msg_input = QLineEdit()
        self.msg_input.returnPressed.connect(self.send_chat_msg)
        layout.addWidget(self.msg_input)
        
        self.send_btn = QPushButton("发送")
        self.send_btn.clicked.connect(self.send_chat_msg)
        layout.addWidget(self.send_btn)
        
        # 复用登录时的socket
        self.socket = None

    def set_socket(self, socket):
        self.socket = socket
        self.socket.readyRead.connect(self.recv_chat_msg)

    def send_chat_msg(self):
        if not self.socket or self.socket.state() != QAbstractSocket.SocketState.ConnectedState:
            return
        msg_text = self.msg_input.text().strip()
        if not msg_text:
            return
        
        # 构造聊天消息JSON
        msg_obj = QJsonObject()
        msg_obj["type"] = "chat"
        msg_obj["content"] = msg_text
        doc = QJsonDocument(msg_obj)
        self.socket.write(doc.toJson(QJsonDocument.JsonFormat.Compact) + b"\n")
        self.msg_input.clear()

    def recv_chat_msg(self):
        while self.socket.bytesAvailable() > 0:
            data = self.socket.readLine().trimmed()
            if not data:
                continue
            doc = QJsonDocument.fromJson(data)
            if not doc.isObject():
                return
            msg_obj = doc.object()
            if msg_obj["type"] == "chat":
                self.chat_display.append(f"{msg_obj['sender']}: {msg_obj['content']}")

class LoginWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("登录")
        self.resize(300, 200)
        
        # 登录界面控件
        central_widget = QWidget()
        self.setCentralWidget(central_widget)
        layout = QVBoxLayout(central_widget)
        
        self.username_input = QLineEdit()
        self.username_input.setPlaceholderText("用户名")
        layout.addWidget(self.username_input)
        
        self.password_input = QLineEdit()
        self.password_input.setPlaceholderText("密码")
        self.password_input.setEchoMode(QLineEdit.EchoMode.Password)
        layout.addWidget(self.password_input)
        
        self.login_btn = QPushButton("登录")
        self.login_btn.clicked.connect(self.start_login)
        layout.addWidget(self.login_btn)
        
        # 加载动画控件
        self.loading_label = QLabel()
        self.loading_label.setVisible(False)
        # 替换为你的加载动图路径，比如"loading.gif"
        self.loading_movie = QMovie("loading.gif")
        self.loading_label.setMovie(self.loading_movie)
        layout.addWidget(self.loading_label)
        
        # 初始化Socket
        self.socket = QTcpSocket(self)
        self.socket.connected.connect(self.on_connected)
        self.socket.readyRead.connect(self.on_ready_read)
        self.socket.errorOccurred.connect(self.on_socket_error)
        self.socket.disconnected.connect(self.on_disconnected)

    def start_login(self):
        username = self.username_input.text().strip()
        password = self.password_input.text().strip()
        if not username or not password:
            return
        
        # 禁用登录按钮，显示加载动画
        self.login_btn.setEnabled(False)
        self.loading_label.setVisible(True)
        self.loading_movie.start()
        
        # 非阻塞连接本地服务器（端口根据实际服务器设置，比如8888）
        self.socket.connectToHost("localhost", 8888)

    def on_connected(self):
        # 连接成功后发送登录信息
        login_obj = QJsonObject()
        login_obj["type"] = "login"
        login_obj["username"] = self.username_input.text().strip()
        login_obj["password"] = self.password_input.text().strip()
        doc = QJsonDocument(login_obj)
        self.socket.write(doc.toJson(QJsonDocument.JsonFormat.Compact) + b"\n")

    def on_ready_read(self):
        # 读取服务器响应
        data = self.socket.readLine().trimmed()
        if not data:
            return
        doc = QJsonDocument.fromJson(data)
        if not doc.isObject():
            return
        
        response_obj = doc.object()
        if response_obj["type"] == "login_response":
            if response_obj["success"]:
                # 登录成功，切换到聊天窗口
                self.loading_movie.stop()
                self.loading_label.setVisible(False)
                self.chat_window = ChatWindow()
                self.chat_window.set_socket(self.socket)
                self.chat_window.show()
                self.close()
            else:
                # 登录失败，恢复登录按钮
                self.login_btn.setEnabled(True)
                self.loading_movie.stop()
                self.loading_label.setVisible(False)
                print(f"登录失败：{response_obj['msg']}")

    def on_socket_error(self, error):
        # 处理连接错误
        self.login_btn.setEnabled(True)
        self.loading_movie.stop()
        self.loading_label.setVisible(False)
        print(f"连接错误：{self.socket.errorString()}")
        self.socket.close()

    def on_disconnected(self):
        # 处理连接断开
        self.login_btn.setEnabled(True)
        self.loading_movie.stop()
        self.loading_label.setVisible(False)
        print("连接已断开")

if __name__ == "__main__":
    app = QApplication(sys.argv)
    window = LoginWindow()
    window.show()
    sys.exit(app.exec())

关键细节说明 #

1. 非阻塞实现：所有网络操作通过信号触发，不会阻塞UI线程，加载动画可正常播放。
2. 数据格式：用JSON序列化消息，客户端和服务器解析更方便；每个消息末尾加换行符\n，用readLine()读取避免粘包。
3. 加载动画：用QMovie播放GIF动图，登录时显示，连接完成/出错时停止并隐藏。
4. 状态判断：发送消息前检查socket连接状态，避免无效操作。

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