本文目录导读：
在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)
关键要点
-
信号类型：
SIGTERM：系统发送的标准终止信号SIGINT：Ctrl+C触发的中断信号SIGHUP：终端挂起信号
-
最佳实践：
- 设置超时机制,防止无限等待
- 处理二次信号强制退出
- 确保资源正确释放
- 对信号处理保持线程安全
-
注意事项：
- Windows上信号支持有限
- 不要在信号处理器中执行耗时操作
- 考虑使用
threading.Event或asyncio.Event控制关闭
选择哪种方案取决于你的具体场景：单线程、多线程还是异步应用。
