如何不终止daemon线程实现asyncio事件循环的停止与重启

原代码存在两个致命逻辑错误：

• Thread.run()是线程启动后的专属执行入口，当loop.stop()触发run_forever()退出后，整个run()方法执行完毕，后台线程会直接终止。此时直接调用enable()->run()，代码会在**调用enable方法的线程（通常为主线程）**中执行，完全不会回到原daemon线程。
• 跨线程直接操作asyncio事件循环的方法属于非线程安全操作，未使用官方提供的跨线程提交接口，很容易出现状态错乱。

该方案线程和事件循环仅初始化一次，永久驻留后台，禁用时仅暂停监控任务，不终止线程和事件循环，稳定性最高、开销最小，满足不终止线程实现启停的需求。

import asyncio
from threading import Thread

class AsyncMonitor:
    def __init__(self):
        self._thread = None
        self.loop = None
        self._monitor_task = None
        self.enabled = False
        self._daemon = True
        self._thread_name = "AsyncMonitoringService"

    def _thread_run(self):
        # 线程入口，仅在线程启动时执行一次
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop)
        try:
            # 事件循环永久运行，不主动退出
            self.loop.run_forever()
        finally:
            # 循环永久停止时做资源清理
            pending = asyncio.all_tasks(self.loop)
            for task in pending:
                task.cancel()
            self.loop.run_until_complete(self.loop.shutdown_asyncgens())
            self.loop.close()
            self.loop = None

    async def _monitor_coroutine(self):
        # 常驻监控协程
        while self.enabled:
            try:
                # 替换为实际监控逻辑
                await some_async_task_scheduling()
            except Exception as e:
                # 捕获异常避免监控任务意外崩溃
                print(f"监控任务执行异常: {repr(e)}")
            # 执行间隔1秒
            await asyncio.sleep(1)

    def enable(self):
        if self.enabled:
            return
        # 首次启用时启动后台线程，线程仅启动一次
        if not (self._thread and self._thread.is_alive()):
            self._thread = Thread(
                target=self._thread_run,
                name=self._thread_name,
                daemon=self._daemon
            )
            self._thread.start()
        # 等待事件循环初始化完成
        while self.loop is None:
            pass
        self.enabled = True
        # 线程安全提交监控协程到后台事件循环
        self._monitor_task = asyncio.run_coroutine_threadsafe(
            self._monitor_coroutine(), self.loop
        )

    def disable(self):
        if not self.enabled:
            return
        self.enabled = False
        # 线程安全取消运行中的监控任务
        if self._monitor_task and not self._monitor_task.done():
            self.loop.call_soon_threadsafe(self._monitor_task.cancel)

    def stop(self):
        # 永久停止服务，终止循环和线程
        self.disable()
        if self.loop and self.loop.is_running():
            self.loop.call_soon_threadsafe(self.loop.stop)
        if self._thread and self._thread.is_alive():
            self._thread.join()

方案关键点：

• 禁止手动调用线程入口方法：线程启动统一用Thread.start()，入口逻辑单独拆分到_thread_run，避免直接调用导致代码跑到当前线程。
• 所有跨线程操作事件循环的逻辑，统一用asyncio.run_coroutine_threadsafe和loop.call_soon_threadsafe两个线程安全接口提交，避免竞态问题。
• 禁用操作仅取消监控协程，不停止事件循环、不退出线程，下次启用直接重新提交监控任务即可，无需新建类实例。

如果需要每次启用时使用全新的daemon线程运行，无需新建AsyncMonitor实例，可基于方案1调整逻辑，在启用时自动清理旧线程、创建新线程即可：

import asyncio
from threading import Thread

class AsyncMonitor:
    def __init__(self):
        self._thread = None
        self.loop = None
        self._monitor_task = None
        self.enabled = False
        self._daemon = True
        self._thread_name = "AsyncMonitoringService"

    def _thread_run(self):
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop)
        try:
            self.loop.run_forever()
        finally:
            pending = asyncio.all_tasks(self.loop)
            for task in pending:
                task.cancel()
            self.loop.run_until_complete(self.loop.shutdown_asyncgens())
            self.loop.close()
            self.loop = None
            # 线程退出后清空引用
            self._thread = None

    async def _monitor_coroutine(self):
        while self.enabled:
            try:
                await some_async_task_scheduling()
            except Exception as e:
                print(f"监控任务执行异常: {repr(e)}")
            await asyncio.sleep(1)

    def enable(self, use_new_thread: bool = False):
        if self.enabled:
            return
        # 需要新线程或当前无存活线程时，创建新线程
        if use_new_thread or not (self._thread and self._thread.is_alive()):
            # 先清理旧资源
            self.stop()
            self._thread = Thread(
                target=self._thread_run,
                name=self._thread_name,
                daemon=self._daemon
            )
            self._thread.start()
        while self.loop is None:
            pass
        self.enabled = True
        self._monitor_task = asyncio.run_coroutine_threadsafe(
            self._monitor_coroutine(), self.loop
        )

    def disable(self):
        if not self.enabled:
            return
        self.enabled = False
        if self._monitor_task and not self._monitor_task.done() and self.loop:
            self.loop.call_soon_threadsafe(self._monitor_task.cancel)

    def stop(self):
        self.disable()
        if self.loop and self.loop.is_running():
            self.loop.call_soon_threadsafe(self.loop.stop)
        if self._thread and self._thread.is_alive():
            self._thread.join()

使用时调用enable(use_new_thread=True)即可在新的daemon线程中启动监控服务，全程复用同一个AsyncMonitor实例。

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