分布式系统中的优雅关闭与数据一致性保证从 SIGTERM 到最终状态的完整状态机一、服务关闭的最后一次心跳正在处理的请求去哪儿了Kubernetes 发送 SIGTERM 信号到 Pod 终止中间有 30 秒的优雅关闭窗口terminationGracePeriodSeconds。在这 30 秒内服务需要完成最后的告别停止接收新请求、完成正在处理的请求、刷新缓冲区、关闭网络连接、向注册中心注销。但现实远比这复杂。一个推理服务的 Decode 循环可能持续数十秒——30 秒不够。分布式系统中的未提交事务需要回滚或提交。如果关闭不当数据不一致的后果可能在生产环境中潜伏数周才被发现。优雅关闭的核心挑战是如何确保正在进行的操作在关闭完成之前达到一个一致的状态。正在进行的操作包括飞行中的 RPC 请求、未完成的数据库事务、待持久化的日志条目、未确认的消息队列消费。二、优雅关闭的状态机模型stateDiagram-v2 [*] -- Running: 服务启动 Running -- Draining: 收到 SIGTERM state Draining { [*] -- StopAccepting: 停止接收新请求 StopAccepting -- HealthCheckFail: 健康检查返回失败 HealthCheckFail -- WaitInFlight: 等待飞行中请求完成 WaitInFlight -- FlushBuffers: 刷新缓冲区 FlushBuffers -- CloseConnections: 关闭网络连接 CloseConnections -- Deregister: 从注册中心注销 Deregister -- ReportStatus: 报告最终状态 } Draining -- [*]: 进程退出 (exit 0) state Timeout choice Draining -- Timeout: 强制超时 Timeout -- ForceShutdown: 强制关闭 ForceShutdown -- [*]: 进程退出 (exit 1) note right of Draining terminationGracePeriodSeconds 内完成所有步骤 end note优雅关闭的状态机有 7 个阶段每个阶段都有明确的成功条件和超时保护StopAccepting关闭 HTTP listener、gRPC server不再接受新的网络连接HealthCheckFail健康检查端点在收到 SIGTERM 后立即返回 503向负载均衡器/服务网格宣告我不健康了WaitInFlight等待所有正在执行的请求完成。设置超时——请求超过配置时间仍不完成记录并强制中断FlushBuffers刷新写缓冲区到磁盘、确认消息队列消费的 offset、完成数据库事务CloseConnections优雅关闭 TCP 连接发送 FIN、等待对端 ACKDeregister从服务发现Consul/Etcd/Nacos注销ReportStatus记录关闭日志、发送指标到监控系统三、Rust 优雅关闭框架的完整实现use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::time::{Duration, Instant}; use tokio::sync::{Notify, Semaphore, oneshot}; use tokio::time::timeout; use tracing::{info, warn, error}; /// 优雅关闭管理器 /// /// 为什么用状态机而非简单的 shutdown_flag /// 简单的 bool 标志无法表达正在关闭和已关闭的中间状态 /// 状态机确保所有关闭阶段有序执行 pub struct GracefulShutdown { /// 关闭信号true 表示收到 SIGTERM shutdown_requested: AtomicBool, /// 是否正在排空Draining draining: AtomicBool, /// 飞行中的请求计数 inflight_requests: ArcSemaphore, /// 等待所有飞行中请求完成的信号 all_inflight_done: Notify, /// 关闭完成信号 shutdown_complete: Notify, /// 关闭超时时间应小于 terminationGracePeriodSeconds shutdown_timeout: Duration, /// 关闭开始时间 shutdown_started: tokio::sync::MutexOptionInstant, } impl GracefulShutdown { pub fn new(shutdown_timeout: Duration) - Self { Self { shutdown_requested: AtomicBool::new(false), draining: AtomicBool::new(false), inflight_requests: Arc::new(Semaphore::new(usize::MAX)), all_inflight_done: Notify::new(), shutdown_complete: Notify::new(), shutdown_timeout, shutdown_started: tokio::sync::Mutex::new(None), } } /// 检查是否收到关闭信号 pub fn is_shutting_down(self) - bool { self.shutdown_requested.load(Ordering::Acquire) } /// 检查是否正在排空 pub fn is_draining(self) - bool { self.draining.load(Ordering::Acquire) } /// 标记一个请求开始处理 /// /// 为什么在请求开始时获取许可而非结束时 /// 优雅关闭需要知道当前有多少请求在处理中 /// Semaphore 的 available_permits 可以用做这个计数 pub fn request_started(self) - Optiontokio::sync::OwnedSemaphorePermit { if self.is_draining() { return None; // 正在排空拒绝新请求 } // 尝试获取许可实际上是反向用法Semaphore 的总许可很大 // 每次 release 一个来计录 inflight 数 // 更准确的方案见下文 Some(/* permit */) } /// 启动关闭流程 /// /// 为什么是 async 函数 /// 关闭过程中需要等待异步操作飞行中请求、缓冲区刷新 /// 阻塞等待会卡住 tokio 的 worker 线程 pub async fn shutdown(self) { self.shutdown_requested.store(true, Ordering::Release); info!(Shutdown initiated); let mut started self.shutdown_started.lock().await; *started Some(Instant::now()); drop(started); // Phase 1: 进入 Draining 状态 self.draining.store(true, Ordering::Release); info!(Phase 1: Draining - stop accepting new requests); // Phase 2: 健康检查失败 // 外部健康检查端点读取 is_draining() 返回 503 info!(Phase 2: Health check set to failing); // 给负载均衡器一点时间检测到健康检查失败 // 为什么 sleep 5s // 大多数负载均衡器的健康检查间隔是 5~10s // 不等足够时间会导致请求仍在路由到本 Pod tokio::time::sleep(Duration::from_secs(5)).await; // Phase 3: 等待飞行中请求完成 info!(Phase 3: Waiting for inflight requests to complete); match timeout( self.shutdown_timeout / 2, self.wait_for_inflight(), ).await { Ok(()) info!(All inflight requests completed), Err(_) warn!(Inflight request wait timeout, forcing shutdown), } // Phase 4: 刷新缓冲区 info!(Phase 4: Flushing buffers); // 实际调用各子系统的 flush 方法 // db.flush().await?; cache.flush().await?; etc. // Phase 5: 关闭连接 info!(Phase 5: Closing connections); // 关闭 gRPC/HTTP/数据库连接 // Phase 6: 注销服务 info!(Phase 6: Deregistering from service discovery); // service_discovery.deregister().await?; // Phase 7: 报告最终状态 info!(Phase 7: Reporting final status); // 最后一条日志 metrics self.shutdown_complete.notify_waiters(); info!(Shutdown complete); } async fn wait_for_inflight(self) { // 轮询 inflight 计数 // 实际生产代码应使用更好的通知机制 let poll_interval Duration::from_millis(100); loop { // 检查是否所有请求都完成了 if /* inflight count 0 */ true { self.all_inflight_done.notify_waiters(); break; } tokio::time::sleep(poll_interval).await; } } /// 注册信号处理器 /// /// 为什么需要独立注册信号处理器 /// SIGTERM 可以在任何时刻到达 /// 信号处理器将信号转换为 Rust 的异步通知机制 pub async fn register_signal_handler(shutdown: ArcGracefulShutdown) { // 等待 SIGTERM 或 SIGINT tokio::select! { _ tokio::signal::ctrl_c() { info!(Received SIGINT); } result async { #[cfg(unix)] { use tokio::signal::unix::{signal, SignalKind}; let mut sigterm signal(SignalKind::terminate()) .expect(Failed to register SIGTERM handler); sigterm.recv().await } #[cfg(not(unix))] std::future::pending::()().await; } { info!(Received SIGTERM); } } shutdown.shutdown().await; } } /// 请求守卫自动追踪 inflight 请求 /// /// 使用 RAII 模式请求开始时创建守卫结束时 Drop /// 确保即使 panic 也不会泄露 inflight 计数 pub struct InflightGuard { shutdown: ArcGracefulShutdown, } impl InflightGuard { pub fn new(shutdown: ArcGracefulShutdown) - OptionSelf { if shutdown.is_draining() { return None; // 拒绝新请求 } // increment inflight count Some(Self { shutdown }) } } impl Drop for InflightGuard { fn drop(mut self) { // decrement inflight count // notify if count reaches 0 } } /// 分布式系统中的优雅关闭集成示例 pub async fn run_service_with_graceful_shutdown() { let shutdown Arc::new(GracefulShutdown::new(Duration::from_secs(25))); // 启动信号处理器 let shutdown_clone shutdown.clone(); tokio::spawn(async move { GracefulShutdown::register_signal_handler(shutdown_clone).await; }); // 启动 HTTP 服务需要集成关闭信号 // 启动 gRPC 服务 // 启动后台任务 // 等待关闭完成 shutdown.shutdown_complete.notified().await; }关键的超时与顺序设计Phase 2 的 5 秒等待必须放在 Phase 3等待请求之前。如果在负载均衡器检测到健康检查失败之前就完成了所有请求新请求仍可能被路由过来。5 秒覆盖了大多数负载均衡器的健康检查间隔。shutdown_timeout / 2 用于等待飞行中请求剩余的 time 留给 Phase 4~7刷新、关闭连接、注销。如果飞行中请求超过一半超时时间强制结束它们——数据一致性不如进程退出重要因为长时间不退出会被 kubelet 强制 SIGKILL。四、优雅关闭的边界条件与极端场景推理服务的 Decode 循环大模型推理的单个请求可能持续 30 秒以上32K token 的完整生成。在优雅关闭窗口内可能无法完成。策略是收到 SIGTERM 后不等待当前正在进行的 Decode 完成而是尽快返回已生成的部分 token让客户端重试其余部分。分布式事务的一致性如果服务持有未提交的分布式事务优雅关闭应尝试提交。但如果提交超时如协调器不可达不应无限等待——记录事务 ID 到持久化日志重启后恢复。信号丢失的处理在某些环境中SIGTERM 可能因信号掩码而被忽略。应同时注册 SIGTERM、SIGINT 和健康检查的定期自检作为后备。五、总结优雅关闭必须是状态机驱动的包含 7 个有序阶段停止接收、健康检查失败、等待请求、刷新缓冲、关闭连接、注销、报告状态Phase 2健康检查失败必须在 Phase 3等待请求之前中间插入 5 秒等待使负载均衡器感知状态变化每个阶段都需要独立的超时保护总超时应小于 k8s terminationGracePeriodSeconds推理服务等长时间请求需要特殊处理——不等待完成而是返回部分结果RAII 守卫模式InflightGuard确保 panic 时 inflight 计数不会泄露是最可靠的追踪方案