Python 收到 SIGTERM 后如何退出：完成当前记录再停止

后台批处理正在写结果，收到停止信号就立刻结束，可能留下尚未完成的记录。本文给单进程 Python 脚本加一个停止标记：收到 SIGTERM 或 SIGINT 后，完成已经开始的这条记录、关闭文件，再明确返回停止状态。

示例只生成编号和平方值，方便在自己的工作目录里验证。它适合逐条处理文件、转换数据等同步任务；需要重启后继续处理时，可以再接入SQLite 断点续跑。

1. 方法：在记录边界检查停止标记

处理流程是：

check stop flag → process one item → write → flush → fsync → next check

信号处理函数只保存首次收到的信号编号。文件写入、打印和清理都留在正常执行流程里，避免处理函数打断原来的 I/O 后又调用相同操作。根据 Python signal 官方文档，处理函数在主线程执行，也必须由主线程注册；长时间运行的 C 扩展计算可能延迟处理。

边界约定： 一条已经进入处理阶段的记录可以完成。主循环观察到停止标记后，不再进入下一条记录。停止等待时间取决于当前操作耗时，真实任务需要给网络请求和外部命令设置超时，可以参考子进程超时与日志教程。

2. 准备：下载并运行正常示例

需要 Linux、Python 3.10 或更新版本，无第三方依赖。本文示例在 Linux / Python 3.11.4 下验证；命令可在 bash 或 zsh 中运行。

把 graceful_batch.py 下载到新的工作目录，或者复制下一节的完整代码保存为同名文件。
执行：

python3 --version
python3 graceful_batch.py --help
mkdir graceful-demo
python3 graceful_batch.py graceful-demo/complete.jsonl --items 3 --delay 0
cat graceful-demo/complete.jsonl

程序会打印 DONE completed=3，最后的文件内容应为：

{"id": 1, "square": 1}
{"id": 2, "square": 4}
{"id": 3, "square": 9}

每次运行使用新的输出文件名。"x" 模式保留已有文件；程序不会自动续写或恢复任务。

3. 实现：完整脚本

"""Finish the current JSONL record before stopping. Linux, Python 3.10+."""
import argparse
import json
import math
import os
from pathlib import Path
import signal
import sys
import time

stop_signal = None


def request_stop(signum, frame):
    global stop_signal
    if stop_signal is None:
        stop_signal = signum


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("output", type=Path, help="new JSONL output path")
    parser.add_argument("--items", type=int, default=10)
    parser.add_argument("--delay", type=float, default=1.0,
                        help="demo work seconds per item (0 to 60)")
    args = parser.parse_args()
    if args.items < 1:
        parser.error("--items must be positive")
    if not math.isfinite(args.delay) or not 0 <= args.delay <= 60:
        parser.error("--delay must be finite and between 0 and 60")

    signal.signal(signal.SIGTERM, request_stop)
    signal.signal(signal.SIGINT, request_stop)
    completed = 0
    try:
        with args.output.open("x", encoding="utf-8") as output:
            for item in range(1, args.items + 1):
                if stop_signal is not None:
                    break
                print(f"START item={item}", flush=True)
                time.sleep(args.delay)  # Replace with bounded, synchronous work.
                record = {"id": item, "square": item * item}
                output.write(json.dumps(record) + "\n")
                output.flush()
                os.fsync(output.fileno())
                completed += 1
                print(f"SAVED item={item}", flush=True)
    except OSError as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1

    if stop_signal is not None:
        name = signal.Signals(stop_signal).name
        print(f"STOPPED signal={name} completed={completed}", flush=True)
        return 128 + stop_signal
    print(f"DONE completed={completed}", flush=True)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

将 time.sleep 和平方计算替换为有明确耗时上限的同步工作，保留完成工作之后的保存顺序。按照 os.fsync 官方说明，缓冲文件应先 flush()，再 fsync()。逐条同步会增加开销，适合低吞吐量演示；这不是多文件事务，也不保证任意文件系统上的断电恢复。

time.sleep 文档说明：处理函数未抛出异常时，睡眠被信号打断后会继续等待剩余时间。因此本例收到停止请求后，会完成本条演示等待，而不是立即结束。

退出码由本例明确设定：正常完成为 0，文件错误为 1，参数错误为 2；收到 SIGTERM 后为 143，SIGINT 后为 130。这些停止码不是全部处理成功的标记，也不表示进程未经处理地被信号杀死。

4. 验证：向正在处理的进程发送信号

先做一次手动观察：

python3 graceful_batch.py graceful-demo/manual.jsonl --items 10 --delay 2

看到 START item=1 后按 Ctrl+C。终端应先出现 SAVED item=1，再出现 STOPPED signal=SIGINT completed=1；如果按得较晚，完成数量会增加。

测试 SIGTERM 时，在另一个终端确认该 Python 进程的 PID，然后用实际 PID 替换示例中的 12345：

kill -TERM 12345

为了精确验证“当前记录完成、后续记录未开始”，把下面的代码保存为 verify_stop.py，放在脚本旁边。测试会等待第一条记录开始，再发送信号，并在超时时清理自己的测试子进程。

import json
from pathlib import Path
import selectors
import signal
import subprocess
import sys
import tempfile

script = Path("graceful_batch.py").resolve()
with tempfile.TemporaryDirectory() as work:
    for sig in (signal.SIGTERM, signal.SIGINT):
        target = Path(work) / (sig.name + ".jsonl")
        process = subprocess.Popen(
            [sys.executable, str(script), str(target), "--delay", "2"],
            stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
        )
        try:
            with selectors.DefaultSelector() as ready:
                ready.register(process.stdout, selectors.EVENT_READ)
                assert ready.select(10), "no startup output"
            assert process.stdout.readline().strip() == "START item=1"
            process.send_signal(sig)
            stdout, stderr = process.communicate(timeout=10)
            assert process.returncode == 128 + sig, (stdout, stderr)
            assert "STOPPED" in stdout and "DONE" not in stdout
            rows = [json.loads(line) for line in target.read_text().splitlines()]
            assert rows == [{"id": 1, "square": 1}], rows
            print(f"PASS {sig.name}: saved one record, exit={process.returncode}")
        finally:
            if process.poll() is None:
                process.kill()
                process.communicate()

运行：

python3 verify_stop.py

预期输出：

PASS SIGTERM: saved one record, exit=143
PASS SIGINT: saved one record, exit=130

检查失败时会抛出异常，测试不会继续声称成功。这个验证检查了退出状态与实际 JSONL 内容；临时目录在测试结束时自动删除。

5. 排查：文件错误和停止不及时

已有输出文件或非法参数

逐条运行下面的命令。它们故意失败，不要放进遇错立即退出的 shell 脚本里：

python3 graceful_batch.py graceful-demo/complete.jsonl --delay 0
echo "$?"
python3 graceful_batch.py graceful-demo/invalid.jsonl --delay nan
echo "$?"

第一次应打印文件已存在错误并返回 1，原来的三条记录保持不变。第二次返回 2，且不会创建 invalid.jsonl。输出目录不存在时也会返回 1，先创建自己的目标目录再运行。

请求停止后仍然等待

先检查当前记录是否正在等待网络、子进程或长时间计算。把这些操作拆成较短的工作单元，并设置超时；单纯增加信号处理函数不会给操作提供时间上限。重复 SIGTERM 或 SIGINT 仍然只是请求停止，本例不会因此强制中断当前记录。

使用 kill -9 后没有停止日志

SIGKILL 无法被捕获，此时清理函数和文件关闭流程都没有执行机会。强制终止、磁盘错误或断电仍可能留下不完整的最后一行；不能把这个 JSONL 文件直接当成可靠检查点。

定时任务中的停止状态被当成成功

保留非零退出码，并单独记录“完成”“请求停止”“执行失败”三种结果。入口应固定解释器和工作目录，做法见cron 环境排查教程。本例没有修改服务或调度器配置；它也不会向另行启动的子进程转发信号。

6. 总结

让信号处理函数只记录停止请求，在下一条记录开始前检查它。把当前记录的计算和保存留在正常流程中，并用真实信号、退出码和结果文件验证停止行为。接入实际任务前，为每个工作单元设定耗时上限，并另外设计断点恢复。