Python如何在收到SIGKILL前检测系统重启？

针对启动脚本直接发送SIGKILL导致无法优雅处理的问题，你可以尝试以下几种方法提前检测系统重启/关机事件：

1. 监听systemd关机事件（适用于使用systemd的Linux系统） #

多数现代Linux系统用systemd管理关机流程，系统关机前会创建/run/systemd/shutdown/scheduled文件，你可以用后台线程定时检查这个文件的存在，一旦检测到就立即执行持久化和第三方服务断开逻辑：

import os
import time
from threading import Thread

def handle_shutdown():
    # 替换为你的数据持久化和服务断开逻辑
    print("执行优雅关闭：保存数据并断开第三方服务...")
    # save_collected_data()
    # disconnect_third_party_services()

def monitor_shutdown_signal():
    shutdown_trigger = "/run/systemd/shutdown/scheduled"
    while True:
        if os.path.exists(shutdown_trigger):
            handle_shutdown()
            break
        time.sleep(1)

# 脚本启动时启动监控线程
Thread(target=monitor_shutdown_signal, daemon=True).start()

也可以通过dbus直接监听systemd的关机信号，但需要依赖dbus-python库，实现相对复杂，文件检查的方式更轻量。

2. 脱离启动脚本的进程组 #

如果启动脚本是通过进程组发送SIGKILL（比如kill -KILL -$PARENT_PGID），你可以在Python脚本启动时主动脱离原进程组，避免被误杀：

import os

# 启动时将自身放入新的进程组，脱离父进程的进程组控制
os.setpgrp()

这样启动脚本发送的组信号不会影响到你的脚本，之后你可以正常监听SIGTERM信号，或者用上述的systemd监控方法完成优雅处理。

3. 监控父进程状态 #

你的脚本的父进程是那个启动脚本，你可以定时检查父进程的状态，一旦发现父进程进入停止/僵尸状态，或者直接退出，就立即执行优雅处理：

import os
import psutil
import time
from threading import Thread

def handle_shutdown():
    print("执行优雅关闭：保存数据并断开第三方服务...")
    # save_collected_data()
    # disconnect_third_party_services()

def monitor_parent_process():
    parent_pid = os.getppid()
    while True:
        try:
            parent_proc = psutil.Process(parent_pid)
            # 检查父进程是否处于停止或僵尸状态
            if parent_proc.status() in (psutil.STATUS_STOPPED, psutil.STATUS_ZOMBIE):
                handle_shutdown()
                break
        except psutil.NoSuchProcess:
            # 父进程已退出，触发优雅处理
            handle_shutdown()
            break
        time.sleep(1)

Thread(target=monitor_parent_process, daemon=True).start()

这个方法需要安装psutil库，如果系统允许安装第三方库可以使用。

4. 检查系统运行状态 #

通过系统命令直接查询当前系统是否处于关机/重启状态，比如用systemctl：

import subprocess
import time
from threading import Thread

def handle_shutdown():
    print("执行优雅关闭：保存数据并断开第三方服务...")
    # save_collected_data()
    # disconnect_third_party_services()

def check_system_state():
    while True:
        try:
            # 查询系统运行状态
            result = subprocess.run(
                ["systemctl", "is-system-running"],
                capture_output=True,
                text=True,
                timeout=2
            )
            status = result.stdout.strip()
            if status in ("stopping", "rebooting"):
                handle_shutdown()
                break
        except (subprocess.CalledProcessError, subprocess.TimeoutExpired):
            # 兼容无systemctl的系统，可替换为其他检测逻辑
            pass
        time.sleep(2)

Thread(target=check_system_state, daemon=True).start()

这些方法的优先级推荐：进程组脱离 > systemd文件监控 > 父进程状态监控，具体选择取决于你的系统环境和启动脚本的行为逻辑。

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