Python信号处理全攻略：signal模块用法与实战详解

1. 信号与机制

1.1 理解UNIX信号与Python的调度模型

在 Python 的信号处理场景中，UNIX 信号的概念与 Python 解释器的调度模型是核心。信号属于操作系统层面的事件，通常由内核在特定时刻送达进程。Python 只能在主线程中接收和处理信号，这意味着多线程程序需要在主线程中设置处理逻辑。

要点一：信号处理器的签名应为 def handler(signum, frame):，其中 signum 表示信号编号，frame 提供当前执行帧的上下文信息。此函数应尽量短小，避免在处理器内执行耗时操作。

要点二：常见信号如 SIGINT、SIGTERM、SIGHUP、SIGALRM 等在不同场景下有不同含义。正确的设计是将信号处理放在事件驱动的主循环中，实时记录状态而非在处理器内直接完成复杂逻辑。

import signal, time

def handler(signum, frame):
    print("Received signal:", signum)

signal.signal(signal.SIGINT, handler)
print("等待信号，按 Ctrl+C 触发 SIGINT。")
while True:
    time.sleep(1)

1.2 常见信号及其含义

SIGINT、SIGTERM、SIGHUP是最常用的三类信号：前者用于中断，后者用于请求终止或重新加载配置。理解它们的语义有助于实现可预测的退出策略和热更新能力。

在实际工程中，将 SIGTERM 作为优雅退出的首选信号，并通过信号处理器设置清理钩子，可以确保资源按序释放而非突然中断。

import signal, sys

def term_handler(signum, frame):
    print("收到 SIGTERM，优雅退出...")
    sys.exit(0)

signal.signal(signal.SIGTERM, term_handler)
print("等待 SIGTERM 以进行优雅退出...")
signal.pause()

2. 常用 API 与示例

2.1 注册信号处理器: signal.signal

signal.signal 是最直接的 API，用于将信号对应的处理器绑定到指定信号。通过明确的处理函数，可以实现对不同信号的分支行为，如日志记录、状态切换或资源释放。

在绑定后，主循环中的代码路径应保持健壮性，避免处理器中执行阻塞性操作，以免延迟其他信号的处理。

import signal, time

def handle_sigint(signum, frame):
    print("捕获到 SIGINT，继续执行但进行干预...")

signal.signal(signal.SIGINT, handle_sigint)

print("正在工作，按 Ctrl+C 触发 SIGINT。")
while True:
    time.sleep(1)

2.2 等待与阻塞：alarm、pause、sigwait

alarm 用于在未来某一时刻触发 SIGALRM，适合定时任务的触发场景。pause 会将进程挂起，直到收到信号为止，便于实现简单的事件等待。sigwait 则在阻塞等待指定信号集合时提供了更可控的同步方式，适用于多线程场景。

结合使用时，务必确保信号在等待期间不会打断其他关键代码段，且要注意线程安全与可移植性。

import signal, time

def on_alarm(signum, frame):
    print("SIGALRM 触发")

signal.signal(signal.SIGALRM, on_alarm)
signal.alarm(2)  # 2 秒后触发

print("等待信号或闹钟...")
time.sleep(5)

import signal, time

sigset = {signal.SIGINT, signal.SIGTERM}
signal.pthread_sigmask(signal.SIG_BLOCK, sigset)

print("阻塞 SIGINT 与 SIGTERM，等待其中一个到来...")
signum = signal.sigwait(sigset)
print("接收到信号:", signum)

2.3 退出与清理：SIGHUP/SIGTERM

SIGHUP 常用于通知重新加载配置或日志轮转，SIGTERM 用于请求优雅退出。将这两个信号结合起来，可以实现服务的热更新和平滑停机。

在实现中，对 SIGHUP 的处理应尽量无副作用，做好资源释放与重新加载的分离，以避免在信号处理器中执行冗长操作。

import signal, time, logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def reload_config(signum, frame):
    logger.info("收到 SIGHUP：重新加载配置")
    # 这里放置配置重新加载逻辑

signal.signal(signal.SIGHUP, reload_config)
signal.signal(signal.SIGTERM, lambda s,f: exit(0))

print("服务运行中，等待 SIGHUP 重新加载，SIGTERM 退出。")
while True:
    logger.info("工作中...")
    time.sleep(2)

3. 与事件循环的整合

3.1 与 asyncio 集成的注意事项

在 asyncio 环境下，信号处理需要在主循环中注册。loop.add_signal_handler 可以将信号与回调绑定，确保事件循环能够在信号到来时退出阻塞状态，完成清理或切换状态。

实践中，信号处理回调应简短、非阻塞，将复杂逻辑放在协程中调度，以保持事件循环的高效性。

import asyncio, signal

async def main():
    loop = asyncio.get_running_loop()
    stop = loop.create_future()

    def on_term():
        if not stop.done():
            stop.set_result(True)

    loop.add_signal_handler(signal.SIGINT, on_term)
    loop.add_signal_handler(signal.SIGTERM, on_term)

    print("Asyncio 运行中，收到信号将结束。")
    await stop
    print("收到终止信号，退出。")

asyncio.run(main())

import asyncio, signal

def run():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)

    async def worker():
        while True:
            print("执行工作...")
            await asyncio.sleep(1)

    for s in (signal.SIGINT, signal.SIGTERM):
        loop.add_signal_handler(s, loop.stop)

    loop.run_until_complete(worker())

if __name__ == "__main__":
    run()

3.2 通过 wakeup_fd 实现高效异步处理

set_wakeup_fd 提供了一种把信号事件以字节写入文件描述符的机制，便于将信号事件整合进事件循环或 IO 复用框架中。这是实现高效异步处理的关键技巧之一，尤其在自定义事件循环时尤为有用。

在具体实现中，处理器应尽量短小，主循环通过监听 wakeup_fd 的可读事件来触发后续处理，避免在信号处理上下文执行复杂逻辑。

import signal, os, time

r, w = os.pipe()
signal.set_wakeup_fd(w)

def on_signal(signum, frame):
    # 最小化工作量，实际处理在主循环读取 wakeup_fd 后完成
    pass

signal.signal(signal.SIGTERM, on_signal)

print("主循环中等待信号唤醒 IO 事件。")
while True:
    time.sleep(1)
    # 在真实场景中，这里会通过 select/poll/epoll 等轮询 wakeup_fd 的就绪

4. 实战场景

4.1 优雅降级与退出策略

在服务型应用中，通过捕获 SIGTERM 来触发优雅退出，确保正在处理的任务完成或被正确中止，避免数据不一致或资源泄露。

实现思路通常是设定一个共享变量或未来对象，信号处理器只修改状态，主循环据此执行清理并退出，避免在处理器中执行耗时任务。

import signal, time

terminate = False
def handle_sigterm(signum, frame):
    global terminate
    terminate = True

signal.signal(signal.SIGTERM, handle_sigterm)

print("服务开启，等待 SIGTERM 触发退出。")
while not terminate:
    time.sleep(0.5)

print("已完成清理，退出。")

4.2 日志轮转与重新打开日志

日志轮转是一种典型的运维操作，SIGHUP 常被用于通知日志重新打开，以确保日志写入点在日志分区或轮转时不中断。

通过在 SIGHUP 的处理器中重新初始化日志系统，可以实现无缝切换日志输出目标，减少停机时间。

import logging, signal, time

def reopen_log(signum, frame):
    for h in logging.getLogger().handlers[:]:
        h.close()
        logging.getLogger().removeHandler(h)
    logging.basicConfig(filename='app.log', level=logging.INFO)
    logging.getLogger(__name__).info("日志重新打开完成")

signal.signal(signal.SIGHUP, reopen_log)

logging.basicConfig(filename='app.log', level=logging.INFO)
logging.info("服务启动")
while True:
    time.sleep(2)
    logging.info("心跳")