如何实现Socket客户端监听线程与客户端函数无冲突运行？

现有一个Client类，包含监听线程循环和多个客户端命令方法（如登录、注册、发送消息）。当前存在的问题是：监听线程会在客户端函数等待数据时错误读取数据，导致二者产生数据读取冲突。

1. 锁的覆盖不完整：原代码中read_exception方法未加锁，存在并发读取Socket的风险；且监听循环的空转逻辑会无意义消耗CPU资源。
2. 职责划分模糊：客户端方法的命令响应与服务器主动推送的消息没有明确区分，容易出现监听线程误读命令响应、客户端方法误读推送消息的情况。
3. Socket状态一致性问题：虽然expecting_data标记和Socket阻塞模式切换在锁内执行，但整体逻辑的严谨性不足，可能导致状态不一致。

• 严格保证所有Socket的读写操作都在互斥锁保护下，彻底避免并发访问导致的数据错乱。
• 明确职责划分：客户端方法仅读取对应命令的响应数据，监听线程仅处理服务器主动推送的消息（如其他用户消息、系统通知）。
• 优化监听循环的空转逻辑，添加短暂休眠减少CPU消耗。
• 将监听线程设为守护线程，避免主程序退出时线程残留。

import threading
from functools import wraps
import server_client_data
# 假设socketH是已实现的Socket包装类，Connection是父类
# 若未定义，需自行补充依赖代码

class Client(server_client_data.Connection):
    def __init__(self, host: str, port: int):
        super().__init__(host, port, "Client", None)
        self.socket.connect((host, port))
        self.socket = socketH(self.socket)
        self.logged_in = False
        self.socket.underlying_socket.setblocking(False)
        self.lock = threading.Lock()
        self.expecting_data = False
        self.threads = {}  # 存储监听线程的停止事件

    @staticmethod
    def client_method(func):
        @wraps(func)
        def wrapper(self, *args, **kwargs):
            with self.lock:
                self.expecting_data = True
                # 切换为阻塞模式，确保命令响应能完整读取
                self.socket.underlying_socket.setblocking(True)

                try:
                    result = func(self, *args, **kwargs)
                finally:
                    self.expecting_data = False
                    # 切回非阻塞模式，供监听线程使用
                    self.socket.underlying_socket.setblocking(False)

            return result

        return wrapper

    @client_method
    def login(self, email: str, password: str):
        # 发送登录命令
        self.socket.write_string(server_client_data.SERVER_COMMANDS.login.value)
        self.read_exception()

        self.socket.write_string(email)
        self.socket.write_string(password)

        self.read_exception()

        username = self.socket.read_string()
        user_code = self.socket.read_string()

        self.logged_in = True
        return username, user_code

    @client_method
    def log_out(self):
        self.socket.write_string(server_client_data.SERVER_COMMANDS.logout.value)
        self.read_exception()
        self.logged_in = False

    @client_method
    def sign_up(self, email: str, password: str, username: str):
        self.socket.write_string(server_client_data.SERVER_COMMANDS.sign_up.value)
        self.read_exception()

        self.socket.write_string(email)
        self.socket.write_string(username)
        self.socket.write_string(password)

        self.read_exception()

    @client_method
    def send_message(self, message: str, username: str, user_code: str):
        self.socket.write_string(server_client_data.SERVER_COMMANDS.send_message.value)
        self.read_exception()

        # 检查登录状态（服务器返回的验证）
        self.read_exception()
        self.logged_in = True

        self.socket.write_bool(False)
        self.socket.write_string(username)
        self.socket.write_string(user_code)

        self.read_exception()

        self.socket.write_string("empty test")
        self.socket.write_bool(False)
        self.socket.write_string(message)

        self.read_exception()
        return self.socket.read_string()

    def start_listener(self):
        stop_event = threading.Event()
        # 设置为守护线程，随主程序退出而终止
        thread = threading.Thread(target=self.listener_loop, args=(stop_event,), daemon=True)
        self.threads[stop_event] = thread
        thread.start()

    def listener_loop(self, stop_event):
        while not stop_event.is_set():
            with self.lock:
                if not self.expecting_data:
                    try:
                        # 读取服务器主动推送的消息命令
                        push_command = self.socket.read_string()
                        if push_command:
                            print(f"收到服务器推送命令: {push_command}")
                            # 可扩展：根据推送命令类型执行对应处理逻辑
                            # self.handle_push_command(push_command)
                    except BlockingIOError:
                        # 无数据时跳过
                        pass
                    except Exception as e:
                        print(f"监听循环错误: {e}")
                        break
            # 释放锁后短暂休眠，减少CPU空转消耗
            threading.sleep(0.01)

    def read_exception(self):
        with self.lock:
            code = self.socket.read_byte()
            if code < 0:
                raise ValueError(self.socket.read_string())

1. 补全锁覆盖：给read_exception方法添加锁，确保所有Socket读取操作都在互斥环境中执行。
2. 优化监听循环：添加threading.sleep(0.01)避免空转消耗CPU，同时明确监听线程仅处理主动推送消息。
3. 守护线程设置：启动监听线程时标记为守护线程，避免主程序退出后线程残留。
4. 职责明确化：客户端方法仅处理自身命令的响应数据，监听线程仅处理服务器主动推送，从逻辑上彻底避免数据读取冲突。

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