Python脚本优雅停机信号如何捕获

本文目录导读：

1. 基础信号处理（最常用）
2. 使用上下文管理器（更Pythonic）
3. 多线程场景
4. 异步IO场景（asyncio）
5. 综合最佳实践
6. 关键要点

在Python中优雅处理停机信号,通常需要捕获 SIGTERM（终止信号）和 SIGINT（Ctrl+C中断信号），以下是几种常用的实现方式：

基础信号处理（最常用）

import signal
import sys
import time
class GracefulShutdown:
    def __init__(self):
        self.shutdown_flag = False
        # 注册信号处理器
        signal.signal(signal.SIGTERM, self.handle_shutdown)
        signal.signal(signal.SIGINT, self.handle_shutdown)
    def handle_shutdown(self, signum, frame):
        print(f"\n收到信号 {signum}，正在优雅关闭...")
        self.shutdown_flag = True
    def run(self):
        while not self.shutdown_flag:
            try:
                # 你的业务逻辑
                print("正在运行...")
                time.sleep(1)
            except Exception as e:
                print(f"运行时错误: {e}")
        self.cleanup()
    def cleanup(self):
        print("执行清理操作...")
        # 关闭数据库连接、文件句柄等
        print("清理完成，退出程序")
        sys.exit(0)
if __name__ == "__main__":
    app = GracefulShutdown()
    app.run()

使用上下文管理器（更Pythonic）

import signal
import sys
import time
from contextlib import contextmanager
@contextmanager
def graceful_shutdown_manager():
    """优雅关闭的上下文管理器"""
    shutdown_event = {'should_stop': False}
    def handler(signum, frame):
        print(f"\n收到信号 {signum}，准备关闭...")
        shutdown_event['should_stop'] = True
    # 保存原信号处理器
    original_sigterm = signal.getsignal(signal.SIGTERM)
    original_sigint = signal.getsignal(signal.SIGINT)
    # 设置新的信号处理器
    signal.signal(signal.SIGTERM, handler)
    signal.signal(signal.SIGINT, handler)
    try:
        yield shutdown_event
    finally:
        # 恢复原信号处理器
        signal.signal(signal.SIGTERM, original_sigterm)
        signal.signal(signal.SIGINT, original_sigint)
# 使用示例
class Service:
    def __init__(self):
        self.running = True
    def start(self):
        with graceful_shutdown_manager() as shutdown:
            while not shutdown['should_stop']:
                print("服务运行中...")
                time.sleep(1)
            print("服务正在关闭...")
        print("服务已完全停止")
# if __name__ == "__main__":
#     service = Service()
#     service.start()

多线程场景

import signal
import threading
import time
import sys
class ThreadedGracefulShutdown:
    def __init__(self):
        self.shutdown_event = threading.Event()
        self.worker_threads = []
        # 注册信号
        signal.signal(signal.SIGTERM, self.signal_handler)
        signal.signal(signal.SIGINT, self.signal_handler)
    def signal_handler(self, signum, frame):
        print(f"\n收到关闭信号 {signum}")
        self.shutdown_event.set()
    def worker(self, name):
        while not self.shutdown_event.is_set():
            print(f"工作线程 {name} 运行中...")
            time.sleep(1)
        print(f"工作线程 {name} 已停止")
    def start_workers(self, count=3):
        for i in range(count):
            t = threading.Thread(target=self.worker, args=(f"Worker-{i}",))
            self.worker_threads.append(t)
            t.start()
    def wait_for_shutdown(self):
        """等待所有线程完成"""
        for t in self.worker_threads:
            t.join(timeout=5)  # 最多等待5秒
        print("所有线程已完成，程序退出")
        sys.exit(0)
    def run(self):
        self.start_workers()
        self.wait_for_shutdown()
if __name__ == "__main__":
    app = ThreadedGracefulShutdown()
    app.run()

异步IO场景（asyncio）

import asyncio
import signal
import sys
class AsyncGracefulShutdown:
    def __init__(self):
        self.shutdown_event = asyncio.Event()
    async def handle_shutdown(self, signum):
        print(f"\n收到信号 {signum}，正在关闭...")
        self.shutdown_event.set()
    async def cleanup(self):
        print("执行异步清理...")
        await asyncio.sleep(1)  # 模拟清理操作
        print("清理完成")
    async def main_task(self):
        print("异步服务启动...")
        try:
            # 等待关闭信号
            await self.shutdown_event.wait()
        except asyncio.CancelledError:
            pass
        finally:
            await self.cleanup()
    def run(self):
        loop = asyncio.get_event_loop()
        # 设置信号处理器
        for sig in (signal.SIGTERM, signal.SIGINT):
            loop.add_signal_handler(
                sig,
                lambda s=sig: asyncio.create_task(self.handle_shutdown(s))
            )
        try:
            loop.run_until_complete(self.main_task())
        finally:
            loop.close()
if __name__ == "__main__":
    app = AsyncGracefulShutdown()
    app.run()

综合最佳实践

import signal
import sys
import time
import logging
from typing import Callable, Optional
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class GracefulShutdown:
    """完善的优雅关闭实现"""
    def __init__(
        self,
        cleanup_func: Optional[Callable] = None,
        shutdown_timeout: int = 10
    ):
        self.shutdown_flag = False
        self.cleanup_func = cleanup_func
        self.shutdown_timeout = shutdown_timeout
        self._register_signals()
    def _register_signals(self):
        """注册所有需要处理的信号"""
        signals = {
            signal.SIGTERM: 'SIGTERM',
            signal.SIGINT: 'SIGINT',
            signal.SIGHUP: 'SIGHUP',  # 挂起信号
        }
        for sig, name in signals.items():
            try:
                signal.signal(sig, self._signal_handler)
                logger.debug(f"已注册信号 {name}")
            except ValueError:
                # 某些信号在Windows上不可用
                logger.warning(f"无法注册信号 {name}")
    def _signal_handler(self, signum, frame):
        """统一的信号处理器"""
        signal_name = signal.Signals(signum).name
        logger.info(f"收到信号 {signal_name} ({signum})")
        if self.shutdown_flag:
            # 如果已经收到一次关闭信号，强制退出
            logger.warning("收到第二次关闭信号，强制退出")
            sys.exit(1)
        self.shutdown_flag = True
    def run(self, main_func: Callable):
        """运行主函数，支持优雅关闭"""
        try:
            main_func(self)
        except Exception as e:
            logger.error(f"运行时错误: {e}")
            raise
        finally:
            self._final_cleanup()
    def _final_cleanup(self):
        """最终清理"""
        logger.info("开始清理...")
        if self.cleanup_func:
            try:
                self.cleanup_func()
            except Exception as e:
                logger.error(f"清理函数执行失败: {e}")
        logger.info("清理完成")
        sys.exit(0)
# 使用示例
def main_task(app: GracefulShutdown):
    """主业务逻辑"""
    logger.info("程序启动")
    while not app.shutdown_flag:
        try:
            logger.debug("执行任务...")
            time.sleep(1)
        except Exception as e:
            logger.error(f"任务执行失败: {e}")
    logger.info("程序正在关闭...")
def cleanup_resources():
    """清理资源"""
    logger.info("关闭数据库连接...")
    logger.info("保存临时文件...")
    logger.info("释放锁资源...")
# if __name__ == "__main__":
#     app = GracefulShutdown(
#         cleanup_func=cleanup_resources,
#         shutdown_timeout=10
#     )
#     app.run(main_task)

关键要点

1. 信号类型：

   • SIGTERM：系统发送的标准终止信号
   • SIGINT：Ctrl+C触发的中断信号
   • SIGHUP：终端挂起信号

2. 最佳实践：

   • 设置超时机制,防止无限等待
   • 处理二次信号强制退出
   • 确保资源正确释放
   • 对信号处理保持线程安全

3. 注意事项：

   • Windows上信号支持有限
   • 不要在信号处理器中执行耗时操作
   • 考虑使用threading.Event或asyncio.Event控制关闭

选择哪种方案取决于你的具体场景：单线程、多线程还是异步应用。