Asyncio系统信号清理：SIGINT下正确关闭永久运行监视器

嘿，这个场景我经常遇到——用asyncio写的后台监视器要优雅处理Ctrl+C（SIGINT信号），其实核心就是用信号处理器配合循环控制标志，再加上必要的任务清理。我给你完善并修改了代码，让你的Scheduler能正确关闭：

完整实现代码 #

import asyncio
import signal

class Scheduler(object):
    def __init__(self, tasks=None, interval=1):
        self.tasks = tasks or []
        self.interval = interval
        self.running = True  # 控制循环启停的标志
        self.sleep_task = None  # 记录当前的sleep任务，方便快速取消

    async def run_forever(self):
        while self.running:
            # 执行一次所有任务
            await self.run_once()
            print(f'Waiting {self.interval} seconds to make next update...')
            
            # 创建sleep任务并等待，保存引用以便后续取消
            self.sleep_task = asyncio.create_task(asyncio.sleep(self.interval))
            try:
                await self.sleep_task
            except asyncio.CancelledError:
                # 捕获sleep被取消的异常，避免报错
                print("Sleep cancelled, proceeding to shutdown...")
            finally:
                self.sleep_task = None
        
        # 关闭时的清理提示
        print("Scheduler is shutting down gracefully...")

    async def run_once(self):
        # 补全你原有的任务执行逻辑，这里用示例gather执行所有任务
        if self.tasks:
            await asyncio.gather(*[task() for task in self.tasks])

def main():
    # 示例异步任务：你可以替换成自己的业务逻辑
    async def sample_monitor_task():
        print("Executing monitor task at:", asyncio.get_event_loop().time())

    # 初始化调度器，传入任务和间隔时间
    scheduler = Scheduler(tasks=[sample_monitor_task], interval=3)

    # 获取当前事件循环
    loop = asyncio.get_event_loop()

    # 定义SIGINT信号处理器：收到Ctrl+C时触发
    def handle_sigint():
        print("\nReceived SIGINT (Ctrl+C), initiating shutdown...")
        scheduler.running = False
        # 如果当前有sleep任务，直接取消以快速退出等待
        if scheduler.sleep_task and not scheduler.sleep_task.done():
            scheduler.sleep_task.cancel()

    # 给事件循环注册SIGINT处理器
    loop.add_signal_handler(signal.SIGINT, handle_sigint)

    try:
        # 运行调度器直到完成
        loop.run_until_complete(scheduler.run_forever())
    finally:
        # 确保事件循环被正确关闭
        loop.close()
        print("Event loop closed successfully.")

if __name__ == "__main__":
    main()

关键要点解释 #

• 循环控制标志：新增self.running布尔值，替代原有的while True，让我们能通过外部信号安全地终止循环。
• 信号处理器：通过loop.add_signal_handler注册SIGINT的处理函数，收到Ctrl+C时立刻设置self.running = False，并取消当前的sleep任务，避免等待间隔结束才退出。
• 优雅清理：确保当前正在执行的run_once任务会完整执行完毕（不会被中断），只有sleep阶段会被快速终止，保证业务逻辑的完整性。
• 异常处理：捕获asyncio.CancelledError，避免取消sleep任务时抛出未处理的异常。

这样你的监视器在收到Ctrl+C时，会先完成当前正在运行的任务（如果有的话），然后立刻退出等待，最后优雅关闭调度器和事件循环。

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