如何在两秒内捕获两次KeyboardInterrupt后停止asyncio TaskGroup任务？

核心思路 #

• 用asyncio.TaskGroup管理异步任务，给每个任务绑定完成回调，一旦有任务完成就触发整个任务组的取消操作
• 自定义Ctrl+C信号处理逻辑，通过记录上次触发中断的时间，只有2秒内连续触发两次时才终止程序，单次触发仅提示，不影响任务运行

完整代码 #

import asyncio
import time
import signal

last_interrupt_time = 0
INTERVAL_THRESHOLD = 2  # 两次中断的有效间隔（秒）

async def worker(task_id, trigger_event):
    print(f"任务[{task_id}]等待触发事件...")
    await trigger_event.wait()
    print(f"任务[{task_id}]执行完成")
    return task_id

async def main():
    global last_interrupt_time
    trigger_event = asyncio.Event()

    def handle_sigint(_, __):
        nonlocal last_interrupt_time
        now = time.time()
        if now - last_interrupt_time < INTERVAL_THRESHOLD:
            print("\n连续两次中断，程序退出")
            asyncio.get_running_loop().stop()
        else:
            print("\n单次中断，2秒内再按一次才会退出")
            last_interrupt_time = now

    # 注册SIGINT信号处理
    signal.signal(signal.SIGINT, handle_sigint)

    try:
        async with asyncio.TaskGroup() as task_group:
            tasks = []
            for i in range(3):
                task = task_group.create_task(worker(i, trigger_event))
                # 任务完成时取消整个任务组
                def cancel_on_complete(task):
                    if not task_group._is_cancelled():
                        task_group.cancel()
                task.add_done_callback(cancel_on_complete)

            # 模拟5秒后触发事件（实际场景替换为真实触发逻辑）
            await asyncio.sleep(5)
            trigger_event.set()

            await asyncio.gather(*tasks, return_exceptions=True)
    except asyncio.CancelledError:
        print("任务组已取消")
    finally:
        print("程序结束")

if __name__ == "__main__":
    asyncio.run(main())

关键细节解释 #

1. 任务完成触发全局取消：每个任务创建后绑定add_done_callback，任务正常完成时调用task_group.cancel()，TaskGroup会自动取消所有未完成的任务并等待它们结束，完全符合“任意任务完成就取消所有”的需求
2. 两次Ctrl+C退出逻辑：通过signal.signal接管SIGINT信号（即Ctrl+C），第一次触发时记录当前时间并提示用户，第二次如果在2秒内触发，直接终止事件循环；单次触发不会终止程序，无需重新创建任务
3. 异常处理：捕获asyncio.CancelledError处理任务组被取消的情况，确保程序优雅结束

测试场景 #

• 正常运行：等待5秒后事件触发，任意任务完成后所有任务被取消，程序正常结束
• 单次Ctrl+C：控制台提示“单次中断，2秒内再按一次才会退出”，任务继续运行，无需重启
• 连续两次Ctrl+C：程序立即终止

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