如何将socket设为类变量供后续调用？PyQt聊天应用开发问题

你当前的代码存在以下几类问题，逐一修复如下：

核心错误原因 #

• 语法错误直接原因：Connecter类__init__方法中的with语句末尾缺少冒号，触发语法报错
• 逻辑错误：用with上下文管理socket会在代码块结束后自动关闭socket，后续发送消息时socket已销毁无法使用
• 作用域错误：ChatApp初始化方法中创建的connecter是局部变量，发送消息方法中无法访问
• 类型错误：sendall方法要求传入字节类型，直接传入字符串会触发类型报错
• 连接时序错误：初始化时直接发起连接，若对方还未监听对应端口会直接连接失败，不支持先启动再回连的需求

修复后完整代码 #

from PyQt5.QtCore import *
from PyQt5.QtGui import *
from PyQt5.QtWidgets import *
import socket
import sys

HOST = '127.0.0.1'

class Receiver(QThread):    
    received = pyqtSignal(object)
    connect_success = pyqtSignal()

    def __init__(self, in_port):
        super().__init__()
        self.in_port = in_port
        self.is_running = True

    def run(self):
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            s.bind((HOST, self.in_port))
            s.listen()
            print('Listening...')

            conn, addr = s.accept()
            self.conn = conn
            self.connect_success.emit()
            with conn:
                print('Connected by', addr)
                while self.is_running:
                    try:
                        data = conn.recv(1024)
                        if not data:
                            break
                        self.received.emit(data)
                    except:
                        break

class Connecter():
    def __init__(self, out_port):
        self.out_port = out_port
        self.c = None

    def connect(self):
        # 单独封装连接方法，支持手动触发连接
        if self.c is not None:
            try:
                self.c.close()
            except:
                pass
        self.c = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        try:
            self.c.connect((HOST, self.out_port))
            print("连接成功")
            return True
        except Exception as e:
            print(f"连接失败：{e}")
            self.c = None
            return False

    def send_msg(self, msg):
        if self.c is None:
            return False
        try:
            self.c.sendall(msg.encode('utf-8'))
            return True
        except:
            self.c = None
            return False

class ChatApp(QMainWindow):
    def __init__(self, in_port, out_port):
        super().__init__()
        self.in_port = in_port
        self.out_port = out_port

        # 界面构建
        self.text_area = QTextEdit()
        self.text_area.setReadOnly(True)
        self.message_line = QLineEdit()
        self.message_line.setFocus()
        self.send_btn = QPushButton('Send')
        self.connect_btn = QPushButton('发起连接')

        outerLayout = QVBoxLayout()
        outerLayout.addWidget(self.text_area)
        bottomLayout = QHBoxLayout()
        bottomLayout.addWidget(self.message_line)
        bottomLayout.addWidget(self.send_btn)
        bottomLayout.addWidget(self.connect_btn)
        outerLayout.addLayout(bottomLayout)

        widget = QWidget()
        widget.setLayout(outerLayout)
        self.setWindowTitle(f"Chat App 监听端口:{in_port} 目标端口:{out_port}")
        # 没有对应图标可以注释掉下面这行
        # self.setWindowIcon(QIcon('ChatIcon.png'))
        self.setCentralWidget(widget)

        # 业务逻辑初始化
        self.receiver = Receiver(in_port)
        self.receiver.received.connect(self.add_messages)
        self.connecter = Connecter(out_port)

        # 信号绑定
        self.send_btn.clicked.connect(self.send_message)
        self.message_line.returnPressed.connect(self.send_message)
        self.connect_btn.clicked.connect(self.connect_target)

        # 启动监听线程
        self.receiver.start()

    def connect_target(self):
        if self.connecter.connect():
            self.text_area.append("=== 已连接到对方 ===")
        else:
            self.text_area.append("=== 连接失败，请确认对方已启动监听 ===")

    def send_message(self):
        message = self.message_line.text().strip()
        if not message:
            return
        if self.connecter.send_msg(message):
            self.text_area.append(f"我: {message}")
            self.message_line.clear()
        else:
            self.text_area.append("=== 发送失败，请先连接对方 ===")

    def add_messages(self, message):
        self.text_area.append(f"对方: {message.decode('utf-8')}")

    def closeEvent(self, event):
        self.receiver.is_running = False
        self.receiver.quit()
        self.receiver.wait()
        if self.connecter.c:
            self.connecter.c.close()
        event.accept()

if __name__ == "__main__":
    if len(sys.argv) !=3:
        print("使用方式：python chat.py <监听端口> <目标端口>")
        print("示例：第一个实例运行 python chat.py 8000 9000，第二个实例运行 python chat.py 9000 8000")
        in_port = int(input('请输入监听端口: '))
        out_port = int(input('请输入目标发送端口: '))
    else:
        in_port = int(sys.argv[1])
        out_port = int(sys.argv[2])
    
    app = QApplication(sys.argv)
    myApp = ChatApp(in_port, out_port)
    myApp.show()
    sys.exit(app.exec())

使用说明 #

1. 启动第一个实例，输入监听端口比如8000，目标端口输入9000
2. 启动第二个实例，输入监听端口9000，目标端口输入8000
3. 两个实例都点击「发起连接」按钮，连接成功后即可互发消息
4. 支持先启动两个实例的监听，再手动发起连接，符合需求

内容的提问来源于stack exchange，提问作者Louie Atkins-Turkish