Python中如何运行不阻塞主线程的无限循环？类内API监听场景

这个场景我之前也碰到过，核心就是把阻塞的监听逻辑放到独立线程里，同时保证类内变量的安全读写，下面给你几个落地的方案，附代码示例：

方案1：线程锁保障类变量的安全读写 #

如果你的需求是直接修改类内的变量（比如用列表存储所有API变更），最稳妥的方式是用threading.Lock来保护对这些变量的访问，避免多线程同时读写导致数据混乱。

示例代码：

import threading
import time  # 模拟API事件的等待逻辑

class APIMonitor:
    def __init__(self):
        self.changes = []  # 存储API变更的类变量
        self.lock = threading.Lock()  # 线程锁，保护共享变量
        self.stop_signal = threading.Event()  # 用于优雅停止监听线程
        # 启动后台监听线程
        self.listen_thread = threading.Thread(target=self._listen_for_changes, daemon=True)
        self.listen_thread.start()

    def _listen_for_changes(self):
        """后台线程中的无限监听循环"""
        while not self.stop_signal.is_set():
            # 替换成你实际的API事件获取逻辑
            api_event = self._fetch_api_event()
            if api_event:
                # 加锁后再修改类变量，确保线程安全
                with self.lock:
                    self.changes.append(api_event)
                    print(f"新变更已存储：{api_event}")
            time.sleep(1)  # 模拟API轮询间隔，按需调整

    def _fetch_api_event(self):
        """模拟API返回事件的方法，替换成你的真实API调用"""
        return {"type": "data_update", "content": f"内容{time.time()}"}

    def get_all_changes(self):
        """主线程安全读取变更的方法"""
        with self.lock:
            # 返回副本，避免外部修改影响内部数据
            return self.changes.copy()

    def stop_monitoring(self):
        """优雅停止监听线程"""
        self.stop_signal.set()
        self.listen_thread.join()

# 测试代码
if __name__ == "__main__":
    monitor = APIMonitor()
    # 主线程执行其他业务任务
    for i in range(5):
        print(f"主线程正在处理任务 {i+1}")
        time.sleep(2)
        # 安全读取已存储的变更
        current_changes = monitor.get_all_changes()
        print(f"当前累计变更数量：{len(current_changes)}")
    # 停止监听
    monitor.stop_monitoring()

方案2：用队列解耦数据传递 #

如果不想直接操作共享变量，用queue.Queue来传递事件是更简洁的方式——监听线程把收到的事件放进队列，主线程或其他业务线程从队列取数据处理，队列本身是线程安全的，不用手动加锁。

示例代码：

import threading
import queue
import time

class APIMonitor:
    def __init__(self):
        self.event_queue = queue.Queue()  # 事件传递队列
        self.stop_signal = threading.Event()
        self.listen_thread = threading.Thread(target=self._listen_for_changes, daemon=True)
        self.listen_thread.start()

    def _listen_for_changes(self):
        while not self.stop_signal.is_set():
            api_event = self._fetch_api_event()
            if api_event:
                self.event_queue.put(api_event)  # 将事件放入队列
                print(f"事件已加入队列：{api_event}")
            time.sleep(1)

    def _fetch_api_event(self):
        return {"type": "data_update", "content": f"内容{time.time()}"}

    def get_next_event(self, timeout=None):
        """从队列获取下一个事件，支持超时设置"""
        try:
            return self.event_queue.get(timeout=timeout)
        except queue.Empty:
            return None

    def stop_monitoring(self):
        self.stop_signal.set()
        self.listen_thread.join()

# 测试代码
if __name__ == "__main__":
    monitor = APIMonitor()
    # 主线程处理队列中的事件
    for i in range(5):
        print(f"主线程执行任务 {i+1}")
        event = monitor.get_next_event(timeout=1)
        if event:
            print(f"处理事件：{event}")
        time.sleep(1)
    monitor.stop_monitoring()

关键注意事项 #

• 守护线程设置：daemon=True能让主线程退出时自动终止后台监听线程，避免程序挂起。如果需要等待线程处理完所有数据，可以去掉该设置，最后调用join()等待线程结束。
• 优雅停止：用threading.Event比直接强制杀死线程更安全，能让监听循环完成当前迭代后再退出，避免资源泄漏。
• 数据一致性：如果多个线程都要修改类变量，一定要用锁或队列保证线程安全，否则可能出现数据丢失、重复或逻辑异常。

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