如何在asyncio脚本中通过SIGHUP信号实现优雅停机？

你猜的完全没错——asyncio的信号处理器回调确实是在独立的线程执行器里运行的，所以你在handle_sighup里抛出的异常只会终止那个线程，根本传不到主协程的上下文里，这就是主循环还能继续跑的核心原因。

要实现信号触发的优雅停机，我们得换一种线程安全的思路：用同步原语（比如asyncio.Event）来通知主协程停止，而不是直接抛异常。下面是重构后的可运行代码：

import asyncio
from signal import SIGHUP, SIGINT, SIGTERM

class GracefulShutdownManager:
    def __init__(self):
        self.shutdown_triggered = asyncio.Event()

    def handle_signal(self, signum):
        print(f"Signal {signum} received, starting shutdown...")
        self.shutdown_triggered.set()

    async def run_service(self):
        loop = asyncio.get_running_loop()
        # 可以同时注册多个常见停机信号，统一处理
        for sig in [SIGHUP, SIGINT, SIGTERM]:
            loop.add_signal_handler(sig, self.handle_signal, sig)

        print("Service is running...")
        # 主循环：直到收到停机信号才退出
        while not self.shutdown_triggered.is_set():
            print("running")
            await asyncio.sleep(1)
        
        # 执行优雅停机逻辑：清理资源、关闭连接等
        await self.perform_graceful_shutdown()

    async def perform_graceful_shutdown(self):
        print("Initiating graceful shutdown procedures...")
        # 这里可以替换为实际的资源清理代码：比如关闭数据库连接、等待后台任务结束
        await asyncio.sleep(2)
        print("Graceful shutdown completed successfully.")

if __name__ == "__main__":
    shutdown_manager = GracefulShutdownManager()
    try:
        asyncio.run(shutdown_manager.run_service())
    except asyncio.CancelledError:
        print("Service task was cancelled during shutdown.")

核心逻辑说明： #

• 用asyncio.Event作为线程安全的停机标记：信号处理器只需要调用set()方法，不会触发任何异常，也能跨线程安全通知主协程
• 主协程通过循环检查shutdown_triggered.is_set()，一旦标记为True就退出主循环，进入停机流程
• 支持同时处理多个信号（比如Ctrl+C触发的SIGINT、系统下发的SIGTERM），统一走优雅停机逻辑

进阶优化方案（适配多后台任务场景） #

如果你的服务有多个后台异步任务，建议在停机时主动取消并等待这些任务完成，避免资源泄漏：

async def run_service(self):
    loop = asyncio.get_running_loop()
    for sig in [SIGHUP, SIGINT, SIGTERM]:
        loop.add_signal_handler(sig, self.handle_signal, sig)

    # 启动后台任务
    self.background_task = asyncio.create_task(self.background_worker())

    print("Service is running...")
    while not self.shutdown_triggered.is_set():
        print("running")
        await asyncio.sleep(1)
    
    # 主动取消后台任务并等待其结束
    self.background_task.cancel()
    try:
        await self.background_task
    except asyncio.CancelledError:
        print("Background worker task cancelled.")
    
    await self.perform_graceful_shutdown()

async def background_worker(self):
    while True:
        print("Background worker is processing...")
        await asyncio.sleep(3)

这种方案的优势：

1. 完全避开了跨线程异常传播的坑，逻辑更稳定可控
2. 可以精细化控制停机流程，确保所有资源都被正确清理
3. 适配多信号、多任务的复杂服务场景

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