跨线程调用loop.create_task添加任务随机阻塞问题求助及修复

问题背景 #

• 需求场景：独立线程运行异步协程，主线程处理同步任务
• 异常现象：跨线程调用loop.create_task添加任务时出现随机阻塞，表现时好时坏
• 特殊情况：VSCode调试模式下问题消失，难以复现定位

问题根因 #

asyncio事件循环本身不是线程安全的，直接跨线程调用loop.create_task会触发线程竞争，导致事件循环内部状态混乱。调试时线程调度被断点放缓，竞争概率大幅降低，因此问题暂时隐藏。必须使用官方提供的线程安全接口提交任务。

此外原代码中_run方法内调用loop.close()会直接关闭事件循环，导致后续无法再提交任务，也是潜在问题。

修复后的完整代码 #

import asyncio
import time
from threading import Thread
import threading


class CorotineLoop:
    def __init__(self):
        self.loop = asyncio.new_event_loop()
        self.thread1 = Thread(
            target=self.run_async1,
            daemon=True,
            name="one"
        )
        self.thread1.start()
        self.print_ids()

    def print_ids(self):
        print(f"{threading.current_thread().name}: self: {id(self)}")
        print(f"{threading.current_thread().name}: loop: {id(self.loop)}")

    def run_async1(self):
        self.print_ids()
        asyncio.set_event_loop(self.loop)
        self.loop.run_forever()

    async def _run(self):
        for i in range(2):
            print(f"{threading.current_thread().name}: {i}")
            await asyncio.sleep(1)
        # 若需关闭事件循环，建议单独提供关闭方法，避免在任务中直接关闭

    def submit(self):
        print("submitting...")
        # 使用线程安全的接口提交协程任务
        asyncio.run_coroutine_threadsafe(self._run(), self.loop)
        print("submitted")


cloop = CorotineLoop()
cloop.submit()

print("start waiting..")
while True:
    time.sleep(0.2)

核心修复说明 #

• 线程安全任务提交：用asyncio.run_coroutine_threadsafe替代loop.create_task，该方法会通过线程安全的队列向事件循环传递任务，避免竞争问题
• 封装事件循环：将事件循环从全局变量改为类成员，减少全局状态带来的线程安全风险
• 避免意外关闭循环：移除任务内的loop.close()，如需关闭可单独添加stop_loop方法，通过loop.call_soon_threadsafe(self.loop.stop)线程安全地停止循环

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