Asyncio子进程在中断时无法正确取消的问题排查

问题原因 #

1. asyncio.run()的默认信号处理拦截：asyncio.run()会自动为SIGINT安装默认处理器，收到信号时直接取消main()任务，覆盖了你手动添加的信号处理器，导致graceful_shutdown从未执行，worker任务不会被取消，子进程持续运行直到完成，最终打印"done"。
2. 信号回调的线程不安全操作：直接在信号回调中调用asyncio.create_task存在线程安全问题，信号回调运行在独立的信号处理线程，而create_task必须在事件循环线程中执行。
3. 子进程未主动终止：即使worker任务被取消，子进程不会自动停止，会在后台继续运行（虽非当前问题直接诱因，但属于必要优化）。

修复后的代码 #

import asyncio
import signal

jobs = set()

async def create_worker():
    task = asyncio.create_task(worker())
    jobs.add(task)
    task.add_done_callback(jobs.discard)

async def worker():
    proc = None
    try:
        proc = await asyncio.create_subprocess_shell('sleep 10')
        await proc.communicate()
        print("done")
    except asyncio.CancelledError:
        print("my task is getting cancelled :\\")
        # 终止子进程，避免后台残留
        if proc is not None:
            proc.terminate()
        raise

async def main() -> None:
    loop = asyncio.get_running_loop()
    loop_hold = asyncio.Event()

    async def graceful_shutdown():
        unfinished_jobs = list(jobs)
        # 取消所有worker任务
        for job in unfinished_jobs:
            job.cancel()
        # 等待所有任务完成取消流程
        await asyncio.wait(unfinished_jobs)
        loop_hold.set()

    # 安全注册信号处理器：用call_soon_threadsafe确保在事件循环线程创建任务
    loop.add_signal_handler(
        signal.SIGINT,
        lambda: loop.call_soon_threadsafe(
            asyncio.create_task,
            graceful_shutdown()
        )
    )

    for _ in range(10):
        await create_worker()
    
    await loop_hold.wait()

if __name__ == "__main__":
    # 手动创建并运行事件循环，避免asyncio.run()的默认信号处理
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        loop.run_until_complete(main())
    finally:
        loop.close()

关键修改说明 #

• 替换asyncio.run()为手动事件循环：完全控制信号处理逻辑，避免默认处理器覆盖自定义逻辑。
• 线程安全的信号回调：使用loop.call_soon_threadsafe将graceful_shutdown任务的创建调度到事件循环线程，解决线程安全问题。
• 主动终止子进程：在worker的CancelledError分支中调用proc.terminate()，确保子进程随任务取消而停止。

内容的提问来源于stack exchange，提问作者brent-mercer