Python多线程类实现TCP通信模拟器通道互扰问题求助

我基于Python 3.12采用面向对象方式开发了一款TCP Socket设备间通信模拟器，SERVER和CLIENT类型通道仅在Socket实例化方式上有差异：SERVER端监听并接受连接请求，CLIENT端主动发起连接。连接建立后，双方通过同一Socket对收发数据、处理响应。

模拟器用Tkinter实现简单界面，网格布局最多支持4个通道，当前测试用2个通道。用户点击CONNECT按钮时，ChannelFrame类的按钮监听器执行以下逻辑：

class ChannelFrame(tk.Frame):
    channel = None # 通道/套接字实例
    
    def connectChannel(self):
        port = self.textPort.get();
        if self.socketType.get() == 'SOCKET_SERVER':
            self.channel = ChannelServerManager(self,self.title,port)
        elif self.socketType.get() == 'SOCKET_CLIENT':
            ipAddress = self.textIP.get()
            self.channel = ChannelClientManager(self,self.title,ipAddress,port)

SERVER和CLIENT通道的构造函数会收集参数并创建主线程，完成以下操作：

• CLIENT端：主动连接对等方
• SERVER端：等待连接请求
• 进入基于select.select的主循环，在对应Frame的文本区域记录收发数据

CLIENT主线程代码 #

class ChannelClientManager():
    establishedConn = None
    receivedData = None
    eventMainThread = None # 用户点击DISCONNECT时触发该事件
    
    def threadClient(self):
        self.socketsInOut.clear()
        self.connected = False
        
        while True:
            if (self.eventMainThread.is_set()):
                print(f"threadClient() --> ChannelClient {self.channelId}: Socket client requested to shut down, exit main loop")
                break;
            
            
            if(not self.connected):
                try :
                    self.establishedConn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                    self.establishedConn.connect((self.ipAddress, int(self.port)))
                    self.channelFrame.setConnectionStateChannel(True)
                    self.socketsInOut.append(self.establishedConn)
                    self.connected = True
                # 持续尝试连接直到成功
                except socket.error as err:
                    print(f'socket.error threadClient() --> ChannelClient {self.channelId}: Error while connecting to server: {err}')
                    time.sleep(0.5)
                    continue
                except socket.timeout as sockTimeout:
                    print(f'socket.timeout threadClient() -->  ChannelClient {self.channelId}: Timeout while connecting to server: {sockTimeout}')
                    continue
                except Exception as e:
                    print(f'Exception on connecting threadClient() -->  ChannelClient {self.channelId}: {e}')
                    continue
                
            
            if(self.connected):
                try:
                    
                    r, _, _ = select.select(self.socketsInOut, [], [], ChannelClientManager.TIMEOUT_SELECT)
                    
                    if len(r) > 0: # 有可读的套接字（收到数据）
                        for fd in r:
                            data = fd.recv(1)
                            if data:
                                self.manageReceivedDataChunk(data)
                            else:
                                print(f"ChannelClient {self.channelId}: Received not data on read socket, server connection closed")
                                self.closeConnection()
                    else:
                        # 超时处理部分数据
                        self.manageReceivedPartialData()
                except ConnectionResetError as crp:
                    print(f"ConnectionResetError threadClient() --> ChannelClient {self.channelId}: {crp}")
                    self.closeConnection()
                except Exception as e:
                    print(f'Exception on selecting threadClient() -->  ChannelClient {self.channelId}: {e}')

SERVER主线程代码 #

class ChannelServerManager():
    socketServer = None # 用于监听/接受连接的套接字
    establishedConn = None # 与对等方建立的连接
    receivedData = None
    eventMainThread = None
    socketsInOut = []

    def __init__(self, channelFrame, channelId, port):
        self.eventMainThread = Event()
        self.socketsInOut.clear()
        self.socketServer = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.socketServer.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.socketServer.bind(('', int(port))) # 监听所有网络接口，若设为127.0.0.1则仅监听回环
        self.socketServer.listen(1) # 仅接受一个客户端连接
        self.socketsInOut.append(self.socketServer)
        
        self.mainThread = Thread(target = self.threadServer)
        self.mainThread.start()


    def threadServer(self): 
            self.receivedData = ''
            
            while True:
                if (self.eventMainThread.is_set()):
                    print("threadServer() --> ChannelServer is requested to shut down, exit main loop")
                    break;
                
                try:
                    r, _, _ = select.select(self.socketsInOut, [], [], ChannelServerManager.TIMEOUT_SELECT)
                    
                    if len(r) > 0: # 有可读的套接字
                        for fd in r:
                            if fd is self.socketServer:
                                
                                # 若监听套接字可读，则有客户端请求连接，接受连接
                                clientsock, clientaddr = self.socketServer.accept()
                                self.establishedConn = clientsock
                                print(f"ChannelServer {self.channelId} is connected from client address {clientaddr}")
                                self.socketsInOut.append(clientsock)
                                self.channelFrame.setConnectionStateChannel(True)
                                self.receivedData = ''
                            elif fd is self.establishedConn:
                                data = fd.recv(1)
                                if not data:
                                    print(f"ChannelServer {self.channelId}: Received not data on read socket, client connection closed")
                                    self.socketsInOut.remove(fd)
                                    self.closeConnection()
                                else:
                                    self.manageReceivedDataChunk(data)
                    else: # 超时处理部分数据
                        self.manageReceivedPartialData()
                except Exception as e:
                    print(f"Exception threadServer() --> ChannelServer {self.channelId}: {traceback.format_exc()}")

遇到的问题 #

多通道/Frame之间出现互相干扰、数据串流的情况：

• 断开某一通道会导致另一通道报错或崩溃
• 医疗设备SERVER仅向端口10001的通道发送数据，部分数据却被另一CLIENT通道接收，日志显示数据分散在Channel 0和Channel 1中

但启动两个单通道的模拟器实例时，一切运行正常，不符合单窗口多通道并行的设计目标。

已尝试的排查步骤 #

最初基于Python ABC库实现ChannelAbstractManager抽象类，让ChannelServerManager和ChannelClientManager继承该类。后来怀疑是Python继承的实例属性共享问题，移除了抽象类，将代码和资源复制到两个类中，但问题依然存在。

求排查思路或解决方案。

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