摘流不是“关掉端口”
8.2 解决了“谁来写”，这一节解决“实例还活着，但该不该继续接流量”。站内 高可用与容错架构 已经讲过探针和故障转移的通用模式，本节只写 arch-lab 订单服务的可运行部分：存活与就绪分离、在途请求计数、优雅停机状态机，以及一组真正启动 HTTP 服务的端到端测试。
直接 kill 进程的问题不是“进程没了”，而是连接被重置、在途订单请求半途失败、上游立刻重试并打到其他实例。摘流要做的三件事是：先让负载均衡停止发送新请求，再等在途请求完成，最后在超时后强制退出。顺序错了，优雅停机就只是“慢一点崩”。
存活与就绪必须分开
存活探针回答“进程是否卡死到需要重启”，就绪探针回答“现在能不能接流量”。两者混用会制造两类事故：把数据库故障写进存活探针，数据库抖动会让所有实例一起重启；把摘流状态只写进存活探针，编排器会在停机时反复重启实例。
arch-lab 的规则是：依赖故障只影响就绪，不影响存活；进入 Draining 后两者都失败，让入口和编排器都看到实例不该再服务。就绪检查可以包含“依赖健康 + 未摘流”，存活检查只包含“进程未进入 Stopped”。
状态机与在途计数
Gate 维护 Serving -> Draining -> Stopped 三态和 inflight 计数。所有业务请求必须经过 Admit：Serving 时计数加一，返回的 done 在请求结束时减一；Draining 和 Stopped 时直接拒绝。Drain 先把状态切到 Draining，再等在途计数归零或超时；超时后仍然进入 Stopped，未完成请求由上游超时兜底。
// file: internal/health/gate.go
// tier: L1
// verify: go test ./internal/health/
package health
import (
"context"
"errors"
"net/http"
"sync"
"time"
)
var ErrDraining = errors.New("draining")
type State int
const (
Serving State = iota
Draining
Stopped
)
func (s State) String() string {
switch s {
case Serving:
return "serving"
case Draining:
return "draining"
default:
return "stopped"
}
}
type Gate struct {
mu sync.Mutex
state State
inflight int
drained chan struct{}
once sync.Once
}
func NewGate() *Gate { return &Gate{drained: make(chan struct{})} }
func (g *Gate) State() State {
g.mu.Lock()
defer g.mu.Unlock()
return g.state
}
func (g *Gate) Inflight() int {
g.mu.Lock()
defer g.mu.Unlock()
return g.inflight
}
func (g *Gate) Admit() (done func(), ok bool) {
g.mu.Lock()
defer g.mu.Unlock()
if g.state != Serving {
return nil, false
}
g.inflight++
return func() {
g.mu.Lock()
defer g.mu.Unlock()
g.inflight--
if g.state == Draining && g.inflight == 0 {
g.closeDrained()
}
}, true
}
func (g *Gate) Drain(timeout time.Duration) bool {
g.mu.Lock()
if g.state != Serving {
g.mu.Unlock()
return false
}
g.state = Draining
if g.inflight == 0 {
g.closeDrained()
}
g.mu.Unlock()
timer := time.NewTimer(timeout)
defer timer.Stop()
select {
case <-g.drained:
g.mu.Lock()
g.state = Stopped
g.mu.Unlock()
return true
case <-timer.C:
g.mu.Lock()
g.state = Stopped
g.mu.Unlock()
return false
}
}
func (g *Gate) closeDrained() { g.once.Do(func() { close(g.drained) }) }
type Checker struct {
Gate *Gate
mu sync.RWMutex
deps map[string]bool
}
func NewChecker(g *Gate) *Checker { return &Checker{Gate: g, deps: map[string]bool{}} }
func (c *Checker) SetDependency(name string, healthy bool) {
c.mu.Lock()
defer c.mu.Unlock()
c.deps[name] = healthy
}
func (c *Checker) Liveness() bool { return c.Gate.State() != Stopped }
func (c *Checker) Readiness() bool {
if c.Gate.State() != Serving {
return false
}
c.mu.RLock()
defer c.mu.RUnlock()
for _, ok := range c.deps {
if !ok {
return false
}
}
return true
}
type Server struct {
Gate *Gate
Checker *Checker
Work func(ctx context.Context, orderID string) error
}
func (s *Server) Handler() http.Handler {
mux := http.NewServeMux()
mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) {
probe(w, s.Checker.Liveness())
})
mux.HandleFunc("/readyz", func(w http.ResponseWriter, _ *http.Request) {
probe(w, s.Checker.Readiness())
})
mux.HandleFunc("/orders", s.handleOrder)
return mux
}
func probe(w http.ResponseWriter, ok bool) {
if !ok {
http.Error(w, "not ready", http.StatusServiceUnavailable)
return
}
w.WriteHeader(http.StatusOK)
_, _ = w.Write([]byte("ok"))
}
func (s *Server) handleOrder(w http.ResponseWriter, r *http.Request) {
done, ok := s.Gate.Admit()
if !ok {
http.Error(w, ErrDraining.Error(), http.StatusServiceUnavailable)
return
}
defer done()
if err := s.Work(r.Context(), r.URL.Query().Get("id")); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
w.WriteHeader(http.StatusOK)
}
httptest 端到端：摘流、在途、超时
第一个测试启动真实 httptest.Server，用阻塞的 Work 把订单请求停在在途状态；随后异步调用 Drain，轮询 /readyz 直到 503；此时新订单请求必须被拒绝，但第一个请求仍能完成，Drain 返回 true 且计数归零。第二个测试让处理器一直阻塞，Drain(50ms) 必须返回 false、状态进入 Stopped、在途计数仍为 1；放行后再确认计数归零。第三个测试验证依赖故障只让就绪失败，不影响存活。
// file: internal/health/gate_test.go
// tier: L1
// verify: go test -v ./internal/health/
package health
import (
"context"
"net/http"
"net/http/httptest"
"testing"
"time"
)
func waitStatus(t *testing.T, url string, want int) {
t.Helper()
deadline := time.Now().Add(2 * time.Second)
for time.Now().Before(deadline) {
resp, err := http.Get(url)
if err != nil {
t.Fatal(err)
}
resp.Body.Close()
if resp.StatusCode == want {
return
}
time.Sleep(5 * time.Millisecond)
}
t.Fatalf("%s never returned %d", url, want)
}
func TestDrainEndToEnd(t *testing.T) {
gate := NewGate()
started, release := make(chan struct{}), make(chan struct{})
srv := httptest.NewServer((&Server{Gate: gate, Checker: NewChecker(gate),
Work: func(_ context.Context, _ string) error { close(started); <-release; return nil }}).Handler())
defer srv.Close()
type result struct {
code int
err error
}
first := make(chan result, 1)
go func() {
resp, err := http.Get(srv.URL + "/orders?id=o-1")
if err != nil {
first <- result{err: err}
return
}
defer resp.Body.Close()
first <- result{code: resp.StatusCode}
}()
<-started
if gate.Inflight() != 1 {
t.Fatalf("inflight before drain=%d", gate.Inflight())
}
drained := make(chan bool, 1)
go func() { drained <- gate.Drain(2 * time.Second) }()
waitStatus(t, srv.URL+"/readyz", http.StatusServiceUnavailable)
resp, err := http.Get(srv.URL + "/orders?id=o-2")
if err != nil {
t.Fatal(err)
}
if resp.StatusCode != http.StatusServiceUnavailable {
t.Fatalf("new request during drain=%d", resp.StatusCode)
}
resp.Body.Close()
close(release)
if got := <-first; got.err != nil || got.code != http.StatusOK {
t.Fatalf("inflight request result=%+v", got)
}
if !<-drained || gate.Inflight() != 0 || gate.State() != Stopped {
t.Fatalf("drain=%v inflight=%d state=%s", true, gate.Inflight(), gate.State())
}
for _, path := range []string{"/orders?id=o-3", "/readyz"} {
resp, err := http.Get(srv.URL + path)
if err != nil {
t.Fatal(err)
}
resp.Body.Close()
if resp.StatusCode != http.StatusServiceUnavailable {
t.Fatalf("%s after stop=%d", path, resp.StatusCode)
}
}
t.Logf("inflight_before_drain=1 drain_graceful=true rejected_new_requests=2 state=%s", gate.State())
}
func TestDrainTimeoutForcesExit(t *testing.T) {
gate := NewGate()
started, release := make(chan struct{}), make(chan struct{})
srv := httptest.NewServer((&Server{Gate: gate, Checker: NewChecker(gate),
Work: func(_ context.Context, _ string) error { close(started); <-release; return nil }}).Handler())
defer srv.Close()
done := make(chan struct{})
go func() {
resp, err := http.Get(srv.URL + "/orders?id=slow-1")
if err == nil {
resp.Body.Close()
}
close(done)
}()
<-started
if gate.Drain(50 * time.Millisecond) {
t.Fatal("drain should time out while request is blocked")
}
if gate.State() != Stopped || gate.Inflight() != 1 {
t.Fatalf("state=%s inflight=%d", gate.State(), gate.Inflight())
}
waitStatus(t, srv.URL+"/healthz", http.StatusServiceUnavailable)
close(release)
<-done
deadline := time.Now().Add(time.Second)
for gate.Inflight() != 0 && time.Now().Before(deadline) {
time.Sleep(5 * time.Millisecond)
}
if gate.Inflight() != 0 {
t.Fatalf("inflight after release=%d", gate.Inflight())
}
t.Logf("drain_graceful=false inflight_at_timeout=1 state=%s inflight_after_release=%d",
gate.State(), gate.Inflight())
}
func TestLivenessReadinessSeparation(t *testing.T) {
checker := NewChecker(NewGate())
checker.SetDependency("postgres", true)
if !checker.Liveness() || !checker.Readiness() {
t.Fatal("healthy instance should be live and ready")
}
checker.SetDependency("postgres", false)
if !checker.Liveness() || checker.Readiness() {
t.Fatal("dependency failure must affect readiness only")
}
t.Logf("liveness_after_dependency_failure=true readiness_after_dependency_failure=false")
}
真实运行结果
以下输出来自本机 macOS（darwin/arm64，Go 1.26.0），命令 go test -v ./... -count=1，运行 1 次；原始输出在 .local/reports/be-arch/be-arch-perf-08-3-health-check-traffic-drain-tests.txt。go test -race ./... 同样通过（ok archlab/internal/health 1.578s）。
=== RUN TestDrainEndToEnd
gate_test.go:81: inflight_before_drain=1 drain_graceful=true rejected_new_requests=2 state=stopped
--- PASS: TestDrainEndToEnd (0.01s)
=== RUN TestDrainTimeoutForcesExit
gate_test.go:115: drain_graceful=false inflight_at_timeout=1 state=stopped inflight_after_release=0
--- PASS: TestDrainTimeoutForcesExit (0.05s)
=== RUN TestLivenessReadinessSeparation
gate_test.go:129: liveness_after_dependency_failure=true readiness_after_dependency_failure=false
--- PASS: TestLivenessReadinessSeparation (0.00s)
PASS
ok archlab/internal/health 0.591s
第一条日志证明摘流后新请求被拒两次（o-2 与停服后的 o-3），在途的 o-1 仍返回 200，计数归零。第二条证明超时后 Drain 返回 false、状态已 Stopped、在途仍为 1；放行后才归零。时间数字是测试里的 50ms 超时和 2 秒等待，不是线上摘流耗时。
探针参数与误摘流的代价
参数是示例起点，必须按业务延迟校准：
| 探针 | 周期 | 超时 | 失败阈值 | 成功阈值 | 动作 |
|---|---|---|---|---|---|
| liveness | 10s | 1s | 3 | 1 | 重启容器 |
| readiness | 2s | 1s | 2 | 1 | 从入口摘除 |
| startup | 2s | 1s | 30 | 1 | 启动期屏蔽前两者 |
就绪失败阈值太小会误摘流：一次 1 秒的垃圾回收或依赖抖动就会让实例退出轮转，流量压到剩余实例，剩余实例的队列变长、超时、又被摘除，形成级联。代价可以用示例值描述：4 实例池摘掉 1 个就损失 25% 容量；如果每个实例还有 20% 预热缓存，摘流后重连和回源会把剩余实例的 P99 推高。真实数字必须用压测和故障注入测，不能引用通用常数。
摘流宽限期也要按在途请求的尾延迟定：示例目标是 P99.9 在途时长 + 探针传播时间，订单创建约 800ms、支付确认约 1.2s，取 30 秒宽限期留足余量；超过宽限期的请求由上游超时和重试兜底。宽限期太短会强制切断正常请求，太长会让发布变慢并延长容量缺口。
常见错误与边界
第一，存活探针检查数据库。数据库抖动会让所有实例同时被重启，故障从依赖层扩散到计算层；依赖只应进入就绪。
第二，先停进程再摘流。入口还没收到摘除信号，新请求仍会进来并直接失败；顺序必须是先切 Draining、等入口生效、再等在途清零。
第三，只在 HTTP 层计数。异步任务、消息消费、数据库事务同样是在途工作；如果它们不在 Admit 的计数里，摘流会在这些任务执行到一半时退出。
第四，宽限期等于平均延迟。必须看尾延迟和最慢请求，平均延迟会掩盖长尾；超时强制退出是兜底，不是常态。
小结
健康检查的价值不在“返回 200”，而在把存活、就绪和摘流分开：依赖故障只摘流量不重启，停机先停止准入再等在途完成，超时才强制退出。本节实现的 Gate 与 Checker 用 httptest 真跑出三组结果：摘流后新请求被拒、在途请求完成、超时强制退出且计数最终归零；探针参数和宽限期都给出了可校准的起点与误摘流代价。下一章进入发布策略，讨论滚动、蓝绿与金丝雀如何复用这套摘流能力。
阅读导航：上一节：8.2 主备与故障转移：单区域高可用 · 下一节：9.1 滚动、蓝绿与金丝雀 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。