如何向asyncio任务发送Ctrl+C信号并在捕获KeyboardInterrupt时终止事件循环中的任务？

当然有靠谱的实现方案！在asyncio的世界里，我们不用直接发送SIGTERM（毕竟这是针对进程的信号），而是用asyncio内置的任务取消机制来实现类似的效果——给任务发送取消信号，让任务能优雅地停止并做清理工作。

核心思路 #

1. 先创建并保存任务的引用，而不是直接把任务函数丢进asyncio.gather，这样后续能主动触发取消操作
2. 在任务内部捕获asyncio.CancelledError，执行必要的清理逻辑（比如关闭连接、保存数据）后重新抛出异常，让asyncio确认任务已被取消
3. 捕获KeyboardInterrupt后，调用每个任务的cancel()方法发送取消信号，再等待任务完成清理

完整代码示例 #

import asyncio

async def task_1():
    try:
        while True:
            print("Task 1 is running...")
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        print("\nTask 1 received cancel signal, cleaning up...")
        # 这里添加任务1的清理逻辑：比如关闭网络连接、临时文件等
        raise  # 必须重新抛出，让asyncio识别任务已被取消

async def task_2():
    try:
        while True:
            print("Task 2 is running...")
            await asyncio.sleep(1.5)
    except asyncio.CancelledError:
        print("\nTask 2 received cancel signal, cleaning up...")
        # 任务2的清理逻辑
        raise

async def main():
    # 创建任务并保存引用，方便后续取消
    task1 = asyncio.create_task(task_1())
    task2 = asyncio.create_task(task_2())
    
    try:
        await asyncio.gather(task1, task2)
    except KeyboardInterrupt:
        print("\nReceived KeyboardInterrupt, starting to cancel tasks...")
        # 给每个任务发送取消信号
        task1.cancel()
        task2.cancel()
        # 等待所有任务完成清理，return_exceptions=True避免因取消异常再次抛出
        await asyncio.gather(task1, task2, return_exceptions=True)
        print("All tasks have been canceled and cleaned up successfully.")

if __name__ == "__main__":
    # 用asyncio.run()管理事件循环，这是Python3.7+推荐的方式
    asyncio.run(main())

关键细节说明 #

• 任务引用的保存：用asyncio.create_task()创建任务并赋值给变量，这样我们能直接调用任务对象的cancel()方法，而不是依赖gather的返回值
• CancelledError的处理：任务内部必须捕获这个异常并重新抛出——如果不重新抛出，asyncio会认为任务是正常结束的，无法正确处理取消状态
• return_exceptions=True：在后续的gather中添加这个参数，能让我们收集所有任务的异常（包括取消异常）而不终止程序，确保所有任务都能完成清理
• 阻塞同步代码的处理：如果你的任务里包含无法await的阻塞同步代码（比如CPU密集型操作），需要把这些代码放到线程池执行（用asyncio.to_thread()或loop.run_in_executor()），这样任务才能响应取消信号。示例如下：

  import time
  async def blocking_task():
      try:
          while True:
              print("Running blocking code...")
              # 把阻塞代码放到线程池，让任务能响应取消
              await asyncio.to_thread(lambda: time.sleep(2))
      except asyncio.CancelledError:
          print("Blocking task canceled, cleaning up...")
          raise
  

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