深度解析:从原理到实战,一文吃透 Linux 信号机制

# 列出所有信号
kill -l

# 输出示例：
#  1) SIGHUP       2) SIGINT       3) SIGQUIT      4) SIGILL
#  5) SIGTRAP      6) SIGABRT      7) SIGBUS       8) SIGFPE
#  9) SIGKILL     10) SIGUSR1     11) SIGSEGV     12) SIGUSR2
# 13) SIGPIPE     14) SIGALRM     15) SIGTERM     16) SIGSTKFLT
# ...

# 第一步：发送 SIGTERM，给程序优雅退出的机会
kill -15 <pid>
# 或
kill <pid>  # 默认就是 -15

# 等待几秒，检查进程是否退出
ps -p <pid>

# 第二步：如果还没退出，再发送 SIGKILL
kill -9 <pid>

# 方式 1：kill 命令（最常用）
kill -信号名 <PID>
kill -SIGTERM 1234
kill -15 1234

# 方式 2：killall（按进程名）
killall -SIGTERM nginx

# 方式 3：pkill（支持正则）
pkill -SIGTERM 'python.*server'

# 方式 4：在代码中发送
kill -SIGUSR1 <pid>  # 自定义信号

import signal
import sys
import time

# 定义信号处理函数
def handle_sigterm(signum, frame):
    print(f"\n收到 SIGTERM 信号 (编号：{signum})")
    print("正在清理资源...")
    # 执行清理操作
    time.sleep(1)
    print("清理完成，退出程序")
    sys.exit(0)

def handle_sigusr1(signum, frame):
    print(f"\n收到 SIGUSR1 信号，执行自定义操作")
    # 自定义逻辑，如重新加载配置
    reload_config()

# 注册信号处理器
signal.signal(signal.SIGTERM, handle_sigterm)
signal.signal(signal.SIGUSR1, handle_sigusr1)

# 主程序
print(f"进程 ID: {os.getpid()}")
print("等待信号... (按 Ctrl+C 发送 SIGINT)")

while True:
    time.sleep(1)
    print(".", end="", flush=True)

const fs = require('fs');

// 优雅退出处理
process.on('SIGTERM', () => {
  console.log('\n收到 SIGTERM 信号');
  console.log('正在关闭服务器...');

  server.close(() => {
    console.log('服务器已关闭');
    process.exit(0);
  });

  // 如果 10 秒后还没关闭，强制退出
  setTimeout(() => {
    console.error('未能优雅关闭，强制退出');
    process.exit(1);
  }, 10000);
});

// 自定义信号处理
process.on('SIGUSR1', () => {
  console.log('收到 SIGUSR1，重新加载配置');
  reloadConfig();
});

console.log(`进程 ID: ${process.pid}`);
console.log('服务运行中...');

import signal
import sys
import time
from http.server import HTTPServer, SimpleHTTPRequestHandler

class GracefulServer:
    def __init__(self, port=8000):
        self.server = HTTPServer(('', port), SimpleHTTPRequestHandler)
        self.shutdown_requested = False

    def handle_sigterm(self, signum, frame):
        print("\n收到退出信号，等待当前请求完成...")
        self.shutdown_requested = True

    def run(self):
        # 注册信号处理器
        signal.signal(signal.SIGTERM, self.handle_sigterm)
        signal.signal(signal.SIGINT, self.handle_sigterm)

        print(f"服务器启动在端口 8000 (PID: {os.getpid()})")

        while not self.shutdown_requested:
            self.server.handle_request()

        print("正在关闭服务器...")
        self.server.server_close()
        print("服务器已关闭")

if __name__ == '__main__':
    server = GracefulServer()
    server.run()

import signal

class TimeoutError(Exception):
    pass

def timeout_handler(signum, frame):
    raise TimeoutError("操作超时!")

def execute_with_timeout(func, timeout_seconds=5, *args, **kwargs):
    """带超时保护的函数执行"""

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

    # 设置定时器
    signal.alarm(timeout_seconds)

    try:
        result = func(*args, **kwargs)
        # 取消定时器
        signal.alarm(0)
        return result
    except TimeoutError as e:
        print(f"错误：{e}")
        return None
    finally:
        # 恢复旧的处理器
        signal.signal(signal.SIGALRM, old_handler)

# 使用示例
def slow_operation():
    time.sleep(10)  # 模拟耗时操作
    return "完成"

result = execute_with_timeout(slow_operation, timeout_seconds=5)
# 5 秒后会抛出 TimeoutError

import signal
import json

class ConfigManager:
    def __init__(self, config_path='config.json'):
        self.config_path = config_path
        self.config = {}
        self.load_config()

        # 注册 SIGHUP 处理器
        signal.signal(signal.SIGHUP, self.reload_config)

    def load_config(self):
        """加载配置文件"""
        with open(self.config_path, 'r') as f:
            self.config = json.load(f)
        print(f"配置已加载：{self.config}")

    def reload_config(self, signum, frame):
        """重载配置（信号处理器）"""
        print("\n收到 SIGHUP 信号，重新加载配置...")
        try:
            self.load_config()
            print("配置重载成功")
        except Exception as e:
            print(f"配置重载失败：{e}")

# 使用方式：
# 1. 启动程序
# 2. 修改 config.json
# 3. 发送信号：kill -HUP <pid>
# 4. 配置自动重载，无需重启

import signal
import os
import time

def handle_sigusr1(signum, frame):
    print(f"[父进程] 收到子进程的 SIGUSR1 信号，任务完成!")

# 父进程
signal.signal(signal.SIGUSR1, handle_sigusr1)

pid = os.fork()

if pid == 0:
    # 子进程
    print(f"[子进程] 开始工作 (PID: {os.getpid()})")
    time.sleep(3)  # 模拟工作
    print("[子进程] 工作完成，发送信号给父进程")

    # 获取父进程 PID 并发送信号
    os.kill(os.getppid(), signal.SIGUSR1)
    os._exit(0)
else:
    # 父进程
    print(f"[父进程] 等待子进程完成 (子进程 PID: {pid})")
    while True:
        time.sleep(1)

# 错误示例
def bad_handler(signum, frame):
    db.connect()  # 可能阻塞
    db.query("SELECT ...")  # 可能死锁
    send_email()  # 可能失败

shutdown_requested = False

def good_handler(signum, frame):
    global shutdown_requested
    shutdown_requested = True  # 只设置标志

while True:
    if shutdown_requested:
        # 在主循环中安全地执行清理
        cleanup()
        break
    do_work()

def handle_exit(signum, frame):
    print("清理资源...")

    # 关闭数据库连接
    db.close()

    # 关闭网络连接
    socket.close()

    # 删除临时文件
    os.remove('/tmp/temp_file')

    # 释放锁
    lock.release()

    sys.exit(0)

import errno

def safe_read(fd, size):
    """处理被信号中断的 read 调用"""
    while True:
        try:
            return os.read(fd, size)
        except OSError as e:
            if e.errno == errno.EINTR:
                # 被信号中断，重试
                continue
            raise

# 1. 查看所有信号
kill -l

# 2. 测试信号处理
python3 signal_test.py &
kill -TERM <pid>

# 3. 实践优雅退出
# 在自己的项目中添加 SIGTERM 处理器

# 4. 尝试热重载
# 实现 SIGHUP 配置重载功能

# 5. 深入学习
# 阅读《Unix 环境高级编程》信号章节