Python多线程类中SocketConnections变量外部无法访问问题

问题根源 #

1. Flask Debug模式的多进程冲突：app.run(debug=True)默认开启自动重载机制，会创建主进程和重载进程两个实例。全局变量global_server_socket在重载进程中会被重新初始化为None，而Socket连接是在原进程的线程中添加的，导致Flask路由访问的是重载进程里的空数组。
2. 线程安全缺失：Socket接收线程与Flask请求线程同时访问SocketConnections列表，无同步机制，可能引发数据可见性异常，路由无法读取最新连接数据。

解决方案 #

1. 关闭Flask自动重载 #

启动Flask时添加use_reloader=False参数，避免多进程导致的全局变量不一致：

修改server.py：

import threading
import time
from flask import Flask, jsonify
import UdpComms as U

app = Flask(__name__)

@app.route('/api/talk', methods=['POST'])
def talk():
    global global_server_socket
    # 加锁安全访问连接列表
    with global_server_socket.lock:
        connections = [
            {"guid": conn.GUID, "address": conn.Address} 
            for conn in global_server_socket.SocketConnections
        ]
    return jsonify(connection_count=len(connections), connections=connections)
    
global_server_socket = None

def start_server():
    global global_server_socket
    sock = U.UdpComms(udpIP="127.0.0.1", portTX=8000, portRX=8001, enableRX=True, suppressWarnings=True)
    global_server_socket = sock
    while True:
        data = sock.ReadReceivedData()
        if data is not None:
            print(data)
        time.sleep(1)

if __name__ == '__main__':
    server_thread = threading.Thread(target=start_server)
    server_thread.start()
    # 关闭自动重载，确保全局变量一致
    app.run(debug=True, host='192.168.0.25', use_reloader=False)

2. 给UdpComms添加线程锁 #

在UdpComms.py中引入线程锁，确保对SocketConnections的所有操作线程安全：

修改后的UdpComms.py：

import json
import uuid 
import socket
import threading

class UdpComms():
    def __init__(self,udpIP,portTX,portRX,enableRX=False,suppressWarnings=True):
        self.SocketConnections = []
        self.lock = threading.Lock()  # 初始化线程锁
        self.udpIP = udpIP
        self.udpSendPort = portTX
        self.udpRcvPort = portRX
        self.enableRX = enableRX
        self.suppressWarnings = suppressWarnings
        self.isDataReceived = False
        self.dataRX = None

        # UDP连接初始化
        self.udpSock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.udpSock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.udpSock.bind((udpIP, portRX))

        # 启动接收线程
        if enableRX:
            self.rxThread = threading.Thread(target=self.ReadUdpThreadFunc, daemon=True)
            self.rxThread.start()

    def __del__(self):
        self.CloseSocket()

    def CloseSocket(self):
        self.udpSock.close()

    def SendData(self, strToSend):
        self.udpSock.sendto(bytes(strToSend,'utf-8'), (self.udpIP, self.udpSendPort))

    def SendDataAddress(self, strToSend, guid):
        print('finding connection: ' + guid)
        with self.lock:  # 加锁访问连接列表
            if self.SocketConnections:
                connection = self.GetConnectionByGUID(guid)
                if connection is not None:
                    print('found connection: ' + guid)
                    self.udpSock.sendto(bytes(strToSend,'utf-8'), connection.Address)

    def ReceiveData(self):
        if not self.enableRX:
            raise ValueError("Attempting to receive data without enabling this setting. Ensure this is enabled from the constructor")

        data = None
        try:
            data, addr = self.udpSock.recvfrom(1024)
            print('Socket data recieved from: ', addr)
            with self.lock:  # 加锁处理新连接
                if self.IsNewConnection(addr):
                    print('New socket')
                    self.SendDataAddress("INIT:" + self.SocketConnections[-1].GUID, self.SocketConnections[-1].GUID)
            data = data.decode('utf-8')
        except WindowsError as e:
            if e.winerror == 10054:
                if not self.suppressWarnings:
                    print("Are You connected to the other application? Connect to it!")
            else:
                raise ValueError("Unexpected Error. Are you sure that the received data can be converted to a string")

        return data

    def ReadUdpThreadFunc(self):
        self.isDataReceived = False
        while True:
            data = self.ReceiveData()
            self.dataRX = data
            self.isDataReceived = True

    def ReadReceivedData(self):
        data = None
        if self.isDataReceived:
            self.isDataReceived = False
            data = self.dataRX
            self.dataRX = None
            if data is not None and data.startswith('DIALOG:'):
                split = data.split(':')[1]
        return data
    
    class Connection:
        def __init__(self, gUID, address) -> None:
            self.GUID = gUID
            self.Address = address

    def IsNewConnection(self, address):
        # 注意：该方法必须在锁保护下调用
        for connection in self.SocketConnections:
            if connection.Address == address:
                return False
        print('Appending new connection...')
        connection = self.Connection(str(uuid.uuid4()), address)
        self.SocketConnections.append(connection)
        return True

    def GetConnectionByGUID(self, guid):
        # 注意：该方法必须在锁保护下调用
        for connection in self.SocketConnections:
            if connection.GUID == guid:
                return connection
        return None

额外说明 #

• 关闭自动重载后，代码修改需手动重启Flask服务，但保证了全局变量的一致性。
• 线程锁避免了多线程并发读写连接列表时的数据异常。
• 路由中需将Connection对象转为字典，因为Flask的jsonify无法直接序列化自定义对象。

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