信号是操作系统用来通知进程发生了某个事件的一种异步通信方式。在Python中，标准库的signal模块提供了处理这些系统信号的机制。信号通常由外部事件触发，例如用户按下Ctrl+C、子进程终止或系统资源耗尽等情况。对于开发系统程序、守护进程或需要长时间运行的应用程序，理解信号处理至关重要。

Python的signal模块作为一个易用的接口，允许开发者定义程序如何响应这些信号，从而构建出健壮、可靠且对外部事件有适当反应的应用程序。信号处理在服务器程序、并行计算和系统工具开发中尤为重要，是Python系统编程的基础知识之一。

基础概念

1、什么是信号

信号本质上是软件中断，代表发送给进程的异步通知。每种信号都有唯一的整数标识符和默认行为。在Unix/Linux系统中，常见的信号包括SIGINT（中断，通常由Ctrl+C触发）、SIGTERM（终止请求）、SIGKILL（强制终止）等。Windows系统支持的信号相对有限，主要包括SIGINT、SIGTERM、SIGABRT和SIGFPE。

信号可以来自多种来源：用户输入、硬件异常、其他进程或内核本身。默认情况下，大多数信号会导致程序终止，但通过signal模块，我们可以修改这个行为，实现自定义的处理逻辑。

2、signal模块的核心功能

Python的signal模块主要提供以下功能：定义信号处理器（回调函数）、发送信号给进程、设置信号阻塞，以及暂停程序执行直到接收到信号。signal模块将系统信号转换为Python事件，允许开发者用Python代码响应这些信号。

信号处理器是当进程接收到特定信号时调用的函数。通过signal.signal()函数，我们可以为特定信号分配自定义处理程序。处理器接收两个参数：信号编号和当前栈帧（通常不使用）。

使用signal模块

1、基本信号处理

使用signal模块最基本的方式是注册一个信号处理器函数，当收到特定信号时执行该函数。下面的示例展示了如何处理SIGINT信号（即用户按下Ctrl+C时触发的信号）。

这个示例创建了一个简单的信号处理程序，使程序在用户按下Ctrl+C时不会立即终止，而是打印一条消息并主动退出。这种模式对于需要在终止前执行清理操作的程序非常有用，比如需要保存数据或释放资源的服务器程序。

import signal
import time
import sys

def signal_handler(sig, frame):
    print('\n您按下了Ctrl+C！程序将优雅地退出。')
    # 在这里执行清理操作
    sys.exit(0)

# 注册SIGINT信号的处理器
signal.signal(signal.SIGINT, signal_handler)

print('程序正在运行，按Ctrl+C退出...')
while True:
    # 模拟长时间运行的任务
    time.sleep(1)
    print('.', end='', flush=True)

运行结果：

程序正在运行，按Ctrl+C退出...
........您按下了Ctrl+C！程序将优雅地退出。

2、超时和警报

signal模块的一个重要应用是与alarm函数结合使用，在指定时间后发送SIGALRM信号，实现超时功能。这在需要限制操作时间的场景中非常有用，如网络请求、用户输入等。

下面的示例展示了如何使用SIGALRM信号实现超时功能。代码设置了一个5秒的定时器，如果用户没有在这段时间内输入内容，程序将捕获SIGALRM信号并触发超时处理。注意，SIGALRM在Windows系统上不可用。

import signal
import sys

def timeout_handler(signum, frame):
    print("\n超时！用户没有及时输入。")
    sys.exit(1)

# 注册SIGALRM信号的处理器
signal.signal(signal.SIGALRM, timeout_handler)

# 设置5秒的闹钟
print("请在5秒内输入您的名字：")
signal.alarm(5)

try:
    name = input()
    # 取消闹钟
    signal.alarm(0)
    print(f"你好，{name}!")
except KeyboardInterrupt:
    print("\n操作被用户中断")

运行结果（如果用户在5秒内输入）：

请在5秒内输入您的名字：
John
你好，John!

运行结果（如果用户没有及时输入）：

请在5秒内输入您的名字：

超时！用户没有及时输入。

3、信号阻塞

在某些情况下，我们需要临时阻止信号处理，确保关键代码段不被信号中断。signal模块提供了signal.pthread_sigmask()函数（仅在Unix系统上可用）来阻塞和解除阻塞信号。

以下示例演示了如何临时阻塞SIGINT信号，确保关键操作不会被用户的Ctrl+C中断。这在需要保证数据完整性的场景中非常重要。

import signal
import time
import os

# 仅在Unix/Linux系统上可用
if hasattr(signal, 'pthread_sigmask'):
    # 定义信号处理器
    def signal_handler(sig, frame):
        print('\n接收到SIGINT信号，但会在关键操作后处理')
        
    # 注册SIGINT处理器
    signal.signal(signal.SIGINT, signal_handler)
    
    print("程序开始运行，随时可按Ctrl+C")
    time.sleep(2)
    
    # 阻塞SIGINT
    print("\n开始关键操作，暂时阻塞SIGINT...")
    old_mask = signal.pthread_sigmask(signal.SIG_BLOCK, [signal.SIGINT])
    
    # 模拟关键操作
    for i in range(5):
        print(f"执行关键操作 {i+1}/5...")
        time.sleep(1)
    
    print("关键操作完成，恢复信号处理")
    # 恢复信号掩码
    signal.pthread_sigmask(signal.SIG_SETMASK, old_mask)
    
    # 给用户时间发送信号
    time.sleep(5)
    print("程序正常结束")

else:
    print("当前系统不支持pthread_sigmask")

运行结果：

程序开始运行，随时可按Ctrl+C

开始关键操作，暂时阻塞SIGINT...
执行关键操作 1/5...
执行关键操作 2/5...
执行关键操作 3/5...
执行关键操作 4/5...
执行关键操作 5/5...
关键操作完成，恢复信号处理
程序正常结束

信号处理的实际应用

1、守护进程和服务程序

下面的示例展示了一个简单的守护进程如何处理各种信号以实现优雅的启动、停止和重新加载配置。这个代码演示了一个简化的守护进程，它能够处理SIGTERM（终止）、SIGHUP（重新加载配置）和SIGUSR1（状态报告）信号。这种模式在开发系统服务、后台任务或需要长时间运行的应用时非常有用。

import signal
import time
import os
import sys
import logging

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    filename='daemon.log'
)
logger = logging.getLogger('daemon')

class SimpleDaemon:
    def __init__(self):
        self.running = False
        self.config = {"interval": 5}  # 模拟配置
    
    def setup_signals(self):
        # 设置信号处理器
        signal.signal(signal.SIGTERM, self.handle_sigterm)
        signal.signal(signal.SIGHUP, self.handle_sighup)
        signal.signal(signal.SIGUSR1, self.handle_sigusr1)
        logger.info("信号处理器已设置")
        
    def handle_sigterm(self, signum, frame):
        """处理终止信号"""
        logger.info("接收到SIGTERM，准备关闭...")
        self.running = False
    
    def handle_sighup(self, signum, frame):
        """处理HUP信号，通常用于重新加载配置"""
        logger.info("接收到SIGHUP，重新加载配置...")
        # 模拟重新加载配置
        self.config = {"interval": 3}
        logger.info(f"配置已更新: {self.config}")
    
    def handle_sigusr1(self, signum, frame):
        """处理USR1信号，用于状态报告"""
        logger.info(f"接收到SIGUSR1，当前状态: 运行中，配置={self.config}")
    
    def run(self):
        """守护进程主循环"""
        self.setup_signals()
        self.running = True
        logger.info(f"守护进程已启动，PID={os.getpid()}")
        
        try:
            while self.running:
                # 执行主要任务
                logger.info(f"执行任务，间隔={self.config['interval']}秒")
                time.sleep(self.config['interval'])
        except Exception as e:
            logger.error(f"发生错误: {e}")
        finally:
            logger.info("守护进程正在关闭...")
            # 执行清理工作
            logger.info("守护进程已关闭")

if __name__ == "__main__":
    daemon = SimpleDaemon()
    daemon.run()

2、优雅地处理多进程应用

以下示例展示了如何在父进程中捕获信号并将其传播给子进程。这个示例创建了一个简单的多进程应用，主进程接收信号并将其传播给所有子进程，确保整个应用可以协调地响应外部事件。这种模式在开发分布式计算系统、Web服务器或其他需要进程池的应用时非常有用。

import signal
import time
import os
import sys
import multiprocessing


def worker_process(worker_id):
    """子进程函数"""

    def handle_signal(signum, frame):
        if signum == signal.SIGTERM:
            print(f"子进程 {worker_id} (PID={os.getpid()}): 接收到终止信号，正在退出...")
            sys.exit(0)
        elif signum == signal.SIGUSR1:
            print(f"子进程 {worker_id} (PID={os.getpid()}): 接收到USR1信号，执行特殊操作...")

    # 子进程设置信号处理器
    signal.signal(signal.SIGTERM, handle_signal)
    signal.signal(signal.SIGUSR1, handle_signal)

    print(f"子进程 {worker_id} (PID={os.getpid()}) 已启动")

    # 模拟工作循环
    while True:  # Fixed syntax error here (missing space)
        time.sleep(1)


def main():
    # 存储子进程对象和PID
    children = []
    child_pids = []
    num_workers = 3

    def parent_signal_handler(signum, frame):
        """父进程信号处理器，负责向子进程传播信号"""
        print(f"父进程 (PID={os.getpid()}): 接收到信号 {signum}，传播给所有子进程...")
        for child_pid in child_pids:
            try:
                os.kill(child_pid, signum)
            except ProcessLookupError:  # More specific exception
                pass  # 子进程可能已经退出

        if signum == signal.SIGTERM:
            # 等待子进程退出
            for child in children:
                child.join(timeout=1)  # Add timeout to prevent hanging
            print("所有子进程已终止，父进程退出")
            sys.exit(0)

    # 设置父进程信号处理器
    signal.signal(signal.SIGTERM, parent_signal_handler)
    signal.signal(signal.SIGUSR1, parent_signal_handler)

    # 创建子进程
    for i in range(num_workers):
        p = multiprocessing.Process(target=worker_process, args=(i,))
        children.append(p)
        p.start()
        child_pids.append(p.pid)

    print(f"父进程 (PID={os.getpid()}) 已启动，子进程: {child_pids}")

    print(f"发送SIGUSR1: kill -USR1 {os.getpid()}")
    print(f"终止程序: kill -TERM {os.getpid()}")

    # 父进程主循环
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("\n接收到键盘中断，正在终止所有进程...")
        parent_signal_handler(signal.SIGTERM, None)


if __name__ == "__main__":
    main()

运行结果：

父进程 (PID=32552) 已启动，子进程: [32555, 32556, 32557]
发送SIGUSR1: kill -USR1 32552
终止程序: kill -TERM 32552
子进程 1 (PID=32556) 已启动
子进程 0 (PID=32555) 已启动
子进程 2 (PID=32557) 已启动

注意事项与最佳实践

1、信号处理的限制

信号处理器中应避免复杂操作。由于信号是异步的，处理器可能在程序执行的任何点被调用，包括其他信号处理或关键操作中，可能导致不可预测的行为。信号处理器应执行简单操作，通常只设置标志变量，让主程序循环检查这些标志并采取适当行动。

下面的代码展示了一种安全的信号处理设计模式，通过设置标志变量而非直接执行复杂逻辑：

import signal
import time
import sys
import os

# 使用全局标志变量
shutdown_requested = False
reload_config_requested = False


def safe_signal_handler(sig, frame):
    """安全的信号处理器，只设置标志，不执行复杂操作"""
    global shutdown_requested, reload_config_requested

    if sig == signal.SIGINT or sig == signal.SIGTERM:
        print("接收到终止信号")
        shutdown_requested = True
    elif sig == signal.SIGHUP:
        print("接收到重载配置信号")
        reload_config_requested = True


def reload_configuration():
    """重载配置的复杂操作"""
    time.sleep(0.5)  # 模拟复杂操作
    print("配置已重载")


def cleanup_resources():
    """清理资源的复杂操作"""
    time.sleep(0.5)  # 模拟复杂操作
    print("资源已清理")


def main_loop():
    """主程序循环，检查标志变量并执行相应操作"""
    global shutdown_requested, reload_config_requested
    
    while not shutdown_requested:
        # 检查是否需要重载配置
        if reload_config_requested:
            print("正在重载配置...")
            # 这里可以安全地执行复杂操作，因为我们在主循环中
            reload_configuration()
            reload_config_requested = False

        # 执行正常工作
        print("执行工作...")
        time.sleep(1)

    print("正在关闭...")
    # 安全地执行清理操作
    cleanup_resources()


if __name__ == "__main__":
    # 设置信号处理器
    signal.signal(signal.SIGINT, safe_signal_handler)
    signal.signal(signal.SIGTERM, safe_signal_handler)
    signal.signal(signal.SIGHUP, safe_signal_handler)

    print(f"程序已启动，PID={os.getpid()}")
    print("使用Ctrl+C或发送SIGTERM终止程序")
    print(f"发送SIGHUP重载配置: kill -HUP {os.getpid()}")
    main_loop()

运行结果：

程序已启动，PID=32682
使用Ctrl+C或发送SIGTERM终止程序
发送SIGHUP重载配置: kill -HUP 32682
执行工作...
接收到重载配置信号
执行工作...
正在重载配置...
配置已重载
执行工作...
接收到终止信号
正在关闭...
资源已清理

信号处理器中应避免非可重入函数（修改全局或静态数据的函数）。安全的函数通常包括基本系统调用和不依赖共享状态的函数。信号不会排队，同一信号在处理器运行时多次发生，系统可能只传递一次。因此，信号适合作为通知机制，而非传输具体数据。

2、跨平台考虑

信号处理在不同操作系统上存在差异。Windows支持有限的信号集（SIGINT、SIGTERM、SIGABRT和SIGFPE），而Unix特有的信号在Windows上不可用。Python的signal模块会在不支持的平台上引发AttributeError。

下面的代码展示了如何编写跨平台兼容的信号处理代码：

import signal
import sys
import time
import platform
import threading  # Moved to top level for consistency


def handle_exit(sig, frame):
    print("接收到退出信号，正在关闭...")
    sys.exit(0)


def setup_signals():
    """根据平台设置可用的信号处理器"""
    # SIGINT在所有平台上都可用
    signal.signal(signal.SIGINT, handle_exit)
    signal.signal(signal.SIGTERM, handle_exit)

    # 仅在Unix平台上设置额外信号
    if platform.system() != "Windows":
        try:
            # 尝试设置Unix特有的信号
            signal.signal(signal.SIGHUP, lambda sig, frame: print("配置重载请求"))
            signal.signal(signal.SIGUSR1, lambda sig, frame: print("收到用户信号1"))
            signal.signal(signal.SIGUSR2, lambda sig, frame: print("收到用户信号2"))
            print("已设置Unix特有的信号处理器")
        except AttributeError:
            print("当前平台不支持某些Unix信号")


def setup_timeout():
    """设置超时处理，考虑平台兼容性"""
    if hasattr(signal, 'SIGALRM'):
        # Unix平台使用SIGALRM实现超时
        def timeout_handler(signum, frame):
            print("操作超时！")
            sys.exit(1)

        signal.signal(signal.SIGALRM, timeout_handler)
        signal.alarm(10)  # 10秒超时
        print("已使用SIGALRM设置超时")
        return True
    else:
        # Windows平台使用替代方案
        print("当前平台不支持SIGALRM，使用线程实现超时")
        timer = threading.Timer(10.0, lambda: (print("操作超时！"), sys.exit(1)))
        timer.start()
        return timer  # 返回timer对象以便后续取消


if __name__ == "__main__":
    print(f"在{platform.system()}平台上运行")
    setup_signals()
    timer_result = setup_timeout()

    try:
        print("程序正在运行，按Ctrl+C退出...")
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        # 处理Ctrl+C（已在signal处理器中处理，但保留此处作为备份）
        pass
    finally:
        # 清理超时计时器（如果是线程）
        if timer_result is not True:  # Only cancel if it's a Timer object
            timer_result.cancel()
        print("程序清理完成")

运行结果：

在Darwin平台上运行
已设置Unix特有的信号处理器
已使用SIGALRM设置超时
程序正在运行，按Ctrl+C退出...
操作超时！
程序清理完成

3、替代方案

某些场景下，其他机制可能比信号更合适。下面的代码比较了使用signal模块和threading.Timer实现超时功能的两种方法：

import time
import sys
import threading
import asyncio


# 方式1：使用threading.Timer实现超时（适用于所有平台）
def timeout_function_threading():
    print("\n使用threading.Timer实现超时示例：")

    def handle_timeout():
        print("操作超时！")
        sys.exit(1)

    # 创建一个3秒后执行handle_timeout的定时器
    timer = threading.Timer(3, handle_timeout)
    timer.start()

    try:
        print("请在3秒内输入内容：")
        user_input = input()
        # 取消定时器
        timer.cancel()
        print(f"你输入了: {user_input}")
    except KeyboardInterrupt:
        timer.cancel()
        print("\n操作被用户取消")
    except Exception as e:
        timer.cancel()
        print(f"发生错误: {e}")


# 方式2：使用asyncio实现更灵活的超时处理
async def get_user_input(timeout):
    try:
        # 创建一个子进程运行系统命令来获取输入
        # 这里使用asyncio的子进程功能，而非阻塞的input()
        process = await asyncio.create_subprocess_exec(
            sys.executable, "-c",
            "import sys; print(sys.stdin.readline(), end='')",
            stdout=asyncio.subprocess.PIPE,
            stdin=asyncio.subprocess.PIPE
        )

        # 设置超时
        try:
            # 需要先发送输入请求提示
            process.stdin.write("请在3秒内输入内容：\n".encode('utf-8'))
            await process.stdin.drain()

            stdout, _ = await asyncio.wait_for(process.communicate(), timeout)
            return stdout.decode().strip()
        except asyncio.TimeoutError:
            # 超时时终止子进程
            process.terminate()
            print("\n输入操作超时！")
            return None
    except Exception as e:
        print(f"发生错误：{e}")
        return None


def timeout_with_asyncio():
    print("\n使用asyncio实现超时示例：")
    try:
        # 运行异步函数
        user_input = asyncio.run(get_user_input(3))
        if user_input:
            print(f"你输入了: {user_input}")
    except ImportError:
        print("此环境不支持asyncio")


# 演示
if __name__ == "__main__":
    print("信号处理替代方案示例")

    # 根据需要选择一种方式演示
    timeout_function_threading()

    print("\n等待5秒后演示下一个方法...")
    time.sleep(5)

    timeout_with_asyncio()

运行结果：

信号处理替代方案示例

使用threading.Timer实现超时示例：
请在3秒内输入内容：
Hello
你输入了: Hello

等待5秒后演示下一个方法...

使用asyncio实现超时示例：
你输入了: 请在3秒内输入内容：

总结

Python的signal模块为操作系统信号处理提供了简洁而强大的接口，使开发者能够开发对外部事件作出响应的健壮应用程序。通过正确使用信号处理，可以实现优雅关闭、配置重载、超时处理等重要功能。信号处理的关键点包括：使用signal.signal()注册信号处理器函数；了解常见信号如SIGINT、SIGTERM和SIGALRM的用途；在多进程应用中协调信号处理；认识到信号处理的限制，如不排队和处理器中应避免复杂操作；以及考虑跨平台兼容性问题。