Python 容器化服务中的并发信号处理与两阶段优雅停机实践

在 Kubernetes 或 Docker 容器编排环境中，容器停止或滚动更新时会先向容器内 1 号进程（PID 1）发送 SIGTERM 信号，若进程未能在宽限期内安全退出，将被强制发送 SIGKILL 终止。本文剖析单线程模型与立即调用 shutdown() 的常见误区，给出基于 ThreadingHTTPServer 的两阶段排空实现，并对比其与生产级 ASGI/WSGI 方案的差异。

一、优雅停机的两个常见工程误区

1. 误区一：使用单线程 HTTPServer 导致探针阻塞
   Python 标准库的 HTTPServer 默认是单线程串行处理请求的。如果某个耗时业务请求（如 /work）正在执行，此时负载均衡器或 Kubernetes 发来的健康检查请求（/healthz）会被阻塞在队列中，无法及时拿到 503 Service Unavailable 响应，甚至触发探针超时误判。因此必须使用多线程并发的 ThreadingHTTPServer。
2. 误区二：收到 SIGTERM 后立即关闭监听套接字
   如果在信号处理函数中立刻调用 server.shutdown()，监听端口会瞬间关闭。但此时 Kubernetes EndpointSlice 同步到 kube-proxy 或 Ingress 网关存在数秒传播延迟，尚未摘除的流量会直接遭遇 Connection refused 报错；同时若未追踪在途请求计数，主线程退出时会导致尚未处理完的在途请求中断。

二、两阶段优雅停机可运行实现 (graceful_server.py)

正确的停机流程应当分为两个阶段：

• 阶段一（探针摘除与流量拒止窗口）：收到 SIGTERM 后立即将健康检查状态置为 503，但保持端口监听一段排空窗口（drain_seconds），让网关探针顺利感知 503 并摘除节点，同时拒绝新流入的业务请求；
• 阶段二（在途请求排空与套接字关闭）：通过 threading.Condition 等待已进入的在途业务请求（active_requests）全部返回 200 OK 并归零后，再调用 server.shutdown() 关闭服务。

完整代码如下：

import json
import signal
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer


class ConcurrentHTTPServer(ThreadingHTTPServer):
    allow_reuse_address = True
    daemon_threads = False


class GracefulService:
    def __init__(self, host="0.0.0.0", port=8080, drain_seconds=5.0, max_wait_seconds=20.0):
        self.shutdown_event = threading.Event()
        self.drain_seconds = drain_seconds
        self.max_wait_seconds = max_wait_seconds
        self.lock = threading.Condition()
        self.active_requests = 0
        self._stopping = False
        self.server = ConcurrentHTTPServer((host, port), self._build_handler())

    def _build_handler(self):
        service = self

        class Handler(BaseHTTPRequestHandler):
            def do_GET(self):
                if self.path == "/healthz":
                    if service.shutdown_event.is_set():
                        self._send_json(503, {"status": "shutting_down"})
                    else:
                        self._send_json(200, {"status": "ok"})
                    return

                if self.path == "/work":
                    with service.lock:
                        if service.shutdown_event.is_set():
                            self._send_json(503, {"error": "server_shutting_down"})
                            return
                        service.active_requests += 1

                    try:
                        time.sleep(2.0)
                        self._send_json(200, {"result": "completed"})
                    finally:
                        with service.lock:
                            service.active_requests -= 1
                            if service.active_requests == 0:
                                service.lock.notify_all()
                    return

                self._send_json(404, {"error": "not_found"})

            def _send_json(self, code, payload):
                body = json.dumps(payload).encode("utf-8")
                self.send_response(code)
                self.send_header("Content-Type", "application/json")
                self.send_header("Content-Length", str(len(body)))
                self.send_header("Connection", "close")
                self.end_headers()
                self.wfile.write(body)

            def log_message(self, fmt, *args):
                print(f"[{self.log_date_time_string()}] {fmt % args}", flush=True)

        return Handler

    def handle_signal(self, signum, _frame):
        with self.lock:
            if self._stopping:
                return
            self._stopping = True
            in_flight = self.active_requests

        sig_name = signal.Signals(signum).name
        print(
            f"收到信号 {sig_name} ({signum}) -> /healthz 切换为 503, 当前在途请求数: {in_flight}",
            flush=True,
        )
        self.shutdown_event.set()
        threading.Thread(target=self._graceful_stop_sequence, daemon=True).start()

    def _graceful_stop_sequence(self):
        time.sleep(self.drain_seconds)
        deadline = time.monotonic() + self.max_wait_seconds
        with self.lock:
            while self.active_requests > 0:
                remaining = deadline - time.monotonic()
                if remaining <= 0:
                    break
                self.lock.wait(timeout=remaining)

        print("排空窗口结束且在途请求已清零，正在停止 HTTPServer...", flush=True)
        self.server.shutdown()

    def serve(self):
        signal.signal(signal.SIGTERM, self.handle_signal)
        signal.signal(signal.SIGINT, self.handle_signal)
        print("服务启动，监听端口 8080...", flush=True)
        self.server.serve_forever()
        self.server.server_close()
        print("服务已安全退出。", flush=True)


if __name__ == "__main__":
    GracefulService().serve()

三、并发与在途排空验证表现

在并发测试及容器 docker stop（发送 SIGTERM）场景下，各请求的实际时序表现如下：

1. 停机前正常请求：访问 /healthz 返回 HTTP 200 {"status": "ok"}；
2. 在途请求与信号并发：当耗时请求 /work 正在执行中时触发 SIGTERM，此时立即并发请求 /healthz，由于 ThreadingHTTPServer 多线程并发处理，探针请求不会被 /work 阻塞，毫秒级返回 HTTP 503 {"status": "shutting_down"}；
3. 停机期新请求拦截：SIGTERM 触发后新发起的 /work 请求立即收到 HTTP 503 {"error": "server_shutting_down"}；
4. 在途请求完整排空：SIGTERM 触发前已进入的在途 /work 请求继续执行完毕并返回 HTTP 200 {"result": "completed"}，随后 active_requests 归零，进程以退出码 0 正常结束，不会触发 docker stop 超时后的 SIGKILL（退出码 137）。

四、标准库简化实现与生产级方案的区别

• 标准库简化方案（本文示例）：基于 ThreadingHTTPServer 与内存级线程锁，适合轻量级内部工具、Sidecar 辅助进程或教学验证，无第三方依赖，但采用每请求一线程模型，不适合高并发长连接场景。
• 生产级微服务方案（FastAPI/Uvicorn/Gunicorn + K8s）：
  1. 使用基于异步事件循环的 Uvicorn（配合 --timeout-graceful-shutdown）或 Gunicorn 多 Worker 进程模型；
  2. 在 Kubernetes Pod 配置中结合 lifecycle.preStop 钩子（如 exec: ["sh", "-c", "sleep 5"]），让 K8s 控制面在向容器发送 SIGTERM 的同时并行更新 EndpointSlice，无需应用层硬编码等待网关传播延迟；
  3. 合理设置 terminationGracePeriodSeconds 大于 preStop 等待时间与最大单次请求耗时之和。