脚本中信号处理怎样优雅退出

本文目录导读：

1. 通用原则
2. Python 实现（最优雅的方式）
3. Shell 脚本中的优雅退出
4. C/C++ 实现（信号安全考量）
5. 关键注意事项
6. 完整示例：网络服务器
7. 优雅退出的核心要素

原子操作、状态清理和避免死锁，以下是一些常见语言和场景的实现策略：

通用原则

# 理想退出流程
1. 设置"退出标志"（原子操作）
2. 主循环检查标志，正常退出
3. 清理资源（文件、网络连接、锁）
4. 退出前发送信号给子进程
5. 恢复终端设置（如果是交互程序）

Python 实现（最优雅的方式）

使用 threading.Event（推荐）

import signal
import threading
import time
# 全局退出事件
exit_event = threading.Event()
def signal_handler(signum, frame):
    print(f"收到信号 {signum}，准备退出...")
    exit_event.set()  # 原子操作，不阻塞
def main_loop():
    signal.signal(signal.SIGINT, signal_handler)
    signal.signal(signal.SIGTERM, signal_handler)
    while not exit_event.is_set():
        # 你的业务逻辑，可以设置超时
        # time.sleep(1)  # 避免忙等待
        # 或者使用 select/poll 等
        # 示例：带超时的等待
        if exit_event.wait(timeout=1):
            break
        do_work()  # 你的实际工作
    cleanup()
def cleanup():
    print("清理资源...")
    # 关闭文件、数据库连接等
    time.sleep(0.5)
    print("安全退出完成")
if __name__ == "__main__":
    main_loop()

异步版本（asyncio）

import asyncio
import signal
class GracefulHandler:
    def __init__(self):
        self.shutdown = False
    def signal_handler(self, signum):
        print(f"收到信号 {signum}，准备优雅退出")
        self.shutdown = True
    async def main(self):
        loop = asyncio.get_event_loop()
        for sig in (signal.SIGINT, signal.SIGTERM):
            loop.add_signal_handler(
                sig, 
                lambda s=sig: self.signal_handler(s)
            )
        while not self.shutdown:
            await asyncio.sleep(1)
            print("工作中...")
        await self.cleanup()
    async def cleanup(self):
        print("清理中...")
        await asyncio.sleep(1)
        print("退出完成")
if __name__ == "__main__":
    app = GracefulHandler()
    try:
        asyncio.run(app.main())
    except KeyboardInterrupt:
        pass

Shell 脚本中的优雅退出

#!/bin/bash
# 全局退出标志
EXIT_FLAG=false
# 信号处理函数
cleanup() {
    echo "收到退出信号，清理资源..."
    EXIT_FLAG=true
    # 杀死子进程
    if [ -n "$CHILD_PID" ]; then
        kill -TERM "$CHILD_PID" 2>/dev/null || true
        wait "$CHILD_PID" 2>/dev/null || true
    fi
    # 清理临时文件
    rm -f /tmp/myapp_*.tmp
    echo "安全退出"
    exit 0
}
# 注册信号处理
trap cleanup SIGINT SIGTERM
# 主循环
while [ "$EXIT_FLAG" = false ]; do
    # 你的业务逻辑
    sleep_handler &
    CHILD_PID=$!
    wait $CHILD_PID 2>/dev/null
    # 检查是否需要退出
    if [ "$EXIT_FLAG" = true ]; then
        break
    fi
done

C/C++ 实现（信号安全考量）

#include <stdio.h>
#include <signal.h>
#include <stdatomic.h>
#include <unistd.h>
// 使用原子变量确保信号安全
static atomic_bool exit_flag = false;
void signal_handler(int signum) {
    // 只设置标志，不做其他操作
    exit_flag = true;
}
int main() {
    struct sigaction sa = {0};
    sa.sa_handler = signal_handler;
    sigaction(SIGINT, &sa, NULL);
    sigaction(SIGTERM, &sa, NULL);
    while (!exit_flag) {
        // 你的业务逻辑
        sleep(1);
        do_work();
    }
    // 清理资源
    cleanup_logs();
    close_database();
    return 0;
}

关键注意事项

信号处理函数必须安全

• 只做最小的操作：设置标志位、写入管道
• 避免：内存分配、锁操作、printf（非信号安全）
• 使用 sig_atomic_t 或 atomic_bool

主循环检查退出条件

# 反模式：纯忙等待
while True:
    if should_exit:
        break
    do_work()  # 如果阻塞，可能永远退不出
# 好模式：使用超时
while not should_exit:
    result = do_work_with_timeout(timeout=1.0)
    if result is None:  # 超时
        continue

资源释放顺序

# 正确的释放顺序（与创建相反）
connections.close()  # 先关连接
files.close()       # 再关文件
locks.release()     # 最后释放锁

子进程处理

import subprocess
import signal
class ProcessManager:
    def __init__(self):
        self.children = []
    def handle_exit(self, signum, frame):
        for p in self.children:
            p.terminate()  # 先发 SIGTERM
        for p in self.children:
            try:
                p.wait(timeout=5)  # 等待最多5秒
            except subprocess.TimeoutExpired:
                p.kill()  # 强制杀死

完整示例：网络服务器

import socket
import select
import signal
import threading
class GracefulServer:
    def __init__(self, host='localhost', port=8080):
        self.running = True
        self.server = socket.socket()
        self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.server.bind((host, port))
        self.server.listen(5)
    def signal_handler(self, signum, frame):
        print(f"收到信号 {signum}")
        self.running = False
    def cleanup(self):
        self.server.close()
        print("服务器已关闭")
    def run(self):
        signal.signal(signal.SIGINT, self.signal_handler)
        signal.signal(signal.SIGTERM, self.signal_handler)
        while self.running:
            # 使用 select 实现超时
            readable, _, _ = select.select(
                [self.server], [], [], 1.0
            )
            if not readable:
                continue  # 超时，重新检查退出标志
            conn, addr = self.server.accept()
            # 处理连接（可以放入线程池）
        self.cleanup()
if __name__ == "__main__":
    server = GracefulServer()
    try:
        server.run()
    except KeyboardInterrupt:
        server.cleanup()

优雅退出的核心要素

优雅退出的目标是让程序在收到信号后，能在合理的时间内安全地结束，不丢失数据、不损坏文件、不留下孤儿进程。