如下的几行代码,作用就是:注册系统信号处理函数,用于在程序被终止时优雅地停止服务(执行清理操作)

# 注册信号处理,用于优雅退出
import signal
signal.signal(signal.SIGTERM, lambda signum, frame: service.stop())
signal.signal(signal.SIGINT, lambda signum, frame: service.stop())

一、import signal

导入 Python 内置的 signal 模块,它提供了与操作系统信号交互的接口
信号是操作系统用来通知进程某些事件发生的一种机制,比如:

  • 用户按下 Ctrl+C(产生 SIGINT
  • 系统要求进程退出(产生 SIGTERM

二、signal.signal(signal.SIGTERM, handler)

注册一个信号处理函数handler)用于响应 SIGTERM 信号。

  • SIGTERM(Signal Terminate)是系统发给程序的终止请求信号,例如执行:

    kill <pid>
    

    就会发送 SIGTERM

  • 注册后,当程序收到 SIGTERM 时,就会自动调用第二个参数指定的函数。

这里注册的处理函数是一个匿名函数(lambda):

lambda signum, frame: service.stop()

这表示当接收到信号时,调用 service.stop() 方法来停止服务释放资源


三、signal.signal(signal.SIGINT, handler)

注册对 SIGINT 信号的处理。

  • SIGINT(Signal Interrupt)是用户在命令行中按下 Ctrl + C 时触发的信号。
  • 注册这个信号意味着用户按下 Ctrl+C 时,程序不会立刻退出,而是先调用 service.stop() 做善后清理。

四、lambda 的参数说明

信号处理函数的标准签名是:

def handler(signum, frame):
    ...
  • signum:信号编号,比如 signal.SIGTERMsignal.SIGINT
  • frame:当前的堆栈帧对象,可用于调试(一般用不到)

在这里虽然 lambda 没有使用这两个参数,但必须写上。


五、总结执行流程

当程序运行时:

  1. 用户按下 Ctrl+C(触发 SIGINT),或外部执行 kill <pid>(触发 SIGTERM);

  2. Python 拦截信号;

  3. 调用 lambda signum, frame: service.stop()

  4. 执行 service.stop(),程序可以:

    • 关闭线程、socket、数据库连接;
    • 保存状态;
    • 然后再退出,达到“优雅退出”的效果。

简化总结一句话:
这几行代码的作用是:

当程序被中断(Ctrl+C)或被系统要求终止(kill)时,自动调用 service.stop(),实现优雅关闭而不是直接强制退出。

附示例代码:

#!/usr/bin/env python3
import time
import logging
from datetime import datetime

logging.basicConfig(
    level=logging.INFO, 
    format='%(asctime)s - %(levelname)s - %(message)s', 
    handlers = [
        logging.FileHandler("/home/momo/myservice/app.log"), 
        logging.StreamHandler()
    ]
)

class MyService:
    def __init__(self):
        self.running = True

    def start(self):
        logging.info("服务启动")
        try:
            while self.running:
                self.perform_task()
                time.sleep(5) # 每5秒执行一次任务
        except Exception as e:
            logging.error(f"服务遇到错误: {e}")
        except KeyboardInterrupt:
            logging.info("服务收到终止信号")
        finally:    
            self.stop()
        

    def perform_task(self):
        current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        logging.info(f"执行任务时间: {current_time}")

    def stop(self):
        if self.running:
            self.running = False
            logging.info("服务停止")


if __name__ == "__main__":
    service = MyService()

    # 捕获终止信号以优雅地停止服务
    import signal
    signal.signal(signal.SIGINT, lambda sig, frame: service.stop())
    signal.signal(signal.SIGTERM, lambda sig, frame: service.stop())

    service.start() 
Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐