系统韧性工程：从被动救火到主动防御的架构演进

Closed → Open: 滑动窗口内失败率超过阈值（如 50%）
Open → Half-Open: 经过冷却时间（如 30s）后自动转换
Half-Open → Closed: 探测请求成功率达标 → 恢复正常
Half-Open → Open: 探测请求仍然失败 → 重新断开

package circuitbreaker

import (
	"errors"
	"sync"
	"time"
)

type State int

const (
	StateClosed State = iota
	StateOpen
	StateHalfOpen
)

var ErrCircuitOpen = errors.New("circuit breaker is open")

type CircuitBreaker struct {
	mu             sync.Mutex
	state          State
	failureThreshold float64
	minRequests     int
	cooldown        time.Duration
	window          []bool // true=success, false=failure
	windowSize      int
	lastFailure    time.Time
	halfOpenMax    int // 半开状态最大探测请求数
	halfOpenCount  int
}

func New(opts ...Option) *CircuitBreaker {
	cb := &CircuitBreaker{
		state:           StateClosed,
		failureThreshold: 0.5,
		minRequests:     20,
		cooldown:        30 * time.Second,
		windowSize:      60,
		halfOpenMax:     5,
	}
	for _, opt := range opts {
		opt(cb)
	}
	return cb
}

// Allow 判断是否允许请求通过
func (cb *CircuitBreaker) Allow() error {
	cb.mu.Lock()
	defer cb.mu.Unlock()

	switch cb.state {
	case StateOpen:
		// 冷却期过了？进入半开状态
		if time.Since(cb.lastFailure) > cb.cooldown {
			cb.state = StateHalfOpen
			cb.halfOpenCount = 0
			return nil // 放行探测请求
		}
		return ErrCircuitOpen
	case StateHalfOpen:
		if cb.halfOpenCount >= cb.halfOpenMax {
			return ErrCircuitOpen // 探测名额用完
		}
		cb.halfOpenCount++
		return nil
	default:
		return nil
	}
}

// Record 记录请求结果
func (cb *CircuitBreaker) Record(success bool) {
	cb.mu.Lock()
	defer cb.mu.Unlock()

	if !success {
		cb.lastFailure = time.Now()
	}

	if cb.state == StateHalfOpen {
		if success {
			cb.state = StateClosed
			cb.window = cb.window[:0] // 清空窗口重新计数
		} else {
			cb.state = StateOpen
		}
		return
	}

	// Closed 状态：记录到滑动窗口
	cb.window = append(cb.window, success)
	if len(cb.window) > cb.windowSize {
		cb.window = cb.window[1:]
	}

	// 检查是否需要熔断
	if len(cb.window) >= cb.minRequests {
		failures := 0
		for _, s := range cb.window {
			if !s {
				failures++
			}
		}
		if float64(failures)/float64(len(cb.window)) >= cb.failureThreshold {
			cb.state = StateOpen
		}
	}
}

type Option func(*CircuitBreaker)

func WithFailureThreshold(t float64) Option {
	return func(cb *CircuitBreaker) { cb.failureThreshold = t }
}

func WithMinRequests(n int) Option {
	return func(cb *CircuitBreaker) { cb.minRequests = n }
}

func WithCooldown(d time.Duration) Option {
	return func(cb *CircuitBreaker) { cb.cooldown = d }
}

cb := circuitbreaker.New(
	circuitbreaker.WithFailureThreshold(0.5),
	circuitbreaker.WithMinRequests(20),
	circuitbreaker.WithCooldown(30*time.Second),
)

func CallPayment(ctx context.Context) (string, error) {
	if err := cb.Allow(); err != nil {
		// 熔断器开着，走降级逻辑
		return "payment_unavailable", nil
	}
	result, err := paymentClient.Charge(ctx)
	cb.Record(err == nil)
	return result, err
}

package bulkhead

import (
	"context"
	"errors"
	"sync"
)

var ErrBulkheadFull = errors.New("bulkhead is full")

type SemaphoreBulkhead struct {
	sem chan struct{}
}

func New(maxConcurrent int) *SemaphoreBulkhead {
	return &SemaphoreBulkhead{
		sem: make(chan struct{}, maxConcurrent),
	}
}

func (b *SemaphoreBulkhead) Acquire(ctx context.Context) error {
	select {
	case b.sem <- struct{}{}:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

func (b *SemaphoreBulkhead) Release() {
	<-b.sem
}

// WithBulkhead 包装函数调用，自动获取和释放信号量
func (b *SemaphoreBulkhead) WithBulkhead(
	ctx context.Context,
	fn func() error,
) error {
	if err := b.Acquire(ctx); err != nil {
		return ErrBulkheadFull
	}
	defer b.Release()
	return fn()
}

package main

import (
	"context"
	"fmt"
	"time"

	"golang.org/x/time/rate"
)

func main() {
	// 每秒 100 个令牌，桶容量 200（允许短暂突发）
	limiter := rate.NewLimiter(rate.Limit(100), 200)

	for i := 0; i < 500; i++ {
		// Wait 会阻塞直到拿到令牌，或 ctx 超时
		if err := limiter.Wait(context.Background()); err != nil {
			fmt.Printf("request %d rejected: %v\n", i, err)
			continue
		}
		fmt.Printf("request %d allowed at %s\n", i, time.Now().Format("15:04:05.000"))
	}
}

用户请求超时:     15s
  → API网关超时:   12s
    → 服务A超时:    10s
      → 服务B超时:  8s
        → 数据库:   5s
        → 缓存:     1s
        → 下游API:  6s

func handleRequest(w http.ResponseWriter, r *http.Request) {
	// 用户请求最多 10 秒
	ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
	defer cancel()

	// 调用下游服务，最多 6 秒
	downstreamCtx, downstreamCancel := context.WithTimeout(ctx, 6*time.Second)
	defer downstreamCancel()

	result, err := downstreamClient.Call(downstreamCtx)
	if err != nil {
		if errors.Is(err, context.DeadlineExceeded) {
			http.Error(w, "service timeout", http.StatusGatewayTimeout)
			return
		}
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	w.Write([]byte(result))
}

func getProductDetail(ctx context.Context, productID string) (*Product, error) {
	product, err := productService.Get(ctx, productID)
	if err != nil {
		// 主路径失败，尝试缓存降级
		cached, cacheErr := cache.Get(ctx, "product:"+productID)
		if cacheErr == nil {
			// 缓存命中，标记为降级数据
			p := decodeProduct(cached)
			p.Degraded = true
			return p, nil
		}
		// 缓存也挂了，返回兜底默认值
		return &Product{
			ID:       productID,
			Name:     "商品信息暂时不可用",
			Degraded: true,
		}, nil
	}
	return product, nil
}

请求入口
  │
  ├─ 第一层：准入控制（白名单/限流）
  │    → 只让合法流量进来，超出速率的直接拒绝
  │
  ├─ 第二层：资源隔离（舱壁隔离）
  │    → 把故障关在小隔间里，不让它扩散
  │
  └─ 第三层：业务兜底（熔断+降级）
       → 最坏情况下给出最小可用结果

package handler

import (
	"context"
	"errors"
	"time"

	"myapp/pkg/bulkhead"
	"myapp/pkg/circuitbreaker"
	"golang.org/x/time/rate"
)

type ResilientHandler struct {
	limiter      *rate.Limiter
	breaker      *circuitbreaker.CircuitBreaker
	bulkhead     *bulkhead.SemaphoreBulkhead
	paymentClient PaymentClient
	fallbackCache FallbackCache
}

func NewResilientHandler() *ResilientHandler {
	return &ResilientHandler{
		// 限流：100 QPS，突发 200
		limiter: rate.NewLimiter(rate.Limit(100), 200),
		// 熔断：失败率 50% 触发，冷却 30s
		breaker: circuitbreaker.New(
			circuitbreaker.WithFailureThreshold(0.5),
			circuitbreaker.WithMinRequests(20),
			circuitbreaker.WithCooldown(30*time.Second),
		),
		// 隔离：最多 50 个并发调用支付服务
		bulkhead: bulkhead.New(50),
	}
}

func (h *ResilientHandler) Charge(ctx context.Context, req *ChargeRequest) (*ChargeResult, error) {
	// 第一层：限流
	if !h.limiter.Allow() {
		return h.fallback(ctx, req, "rate limited")
	}

	// 第二层：熔断检查
	if err := h.breaker.Allow(); err != nil {
		return h.fallback(ctx, req, "circuit open")
	}

	// 第三层：舱壁隔离 + 超时调用
	callCtx, cancel := context.WithTimeout(ctx, 3*time.Second)
	defer cancel()

	var result *ChargeResult
	var callErr error

	err := h.bulkhead.WithBulkhead(callCtx, func() error {
		result, callErr = h.paymentClient.Charge(callCtx, req)
		return callErr
	})

	// 记录熔断器结果
	h.breaker.Record(err == nil)

	if err != nil {
		if errors.Is(err, bulkhead.ErrBulkheadFull) {
			return h.fallback(ctx, req, "bulkhead full")
		}
		return h.fallback(ctx, req, "payment timeout")
	}

	return result, nil
}

// fallback 降级逻辑——必须独立于主路径
func (h *ResilientHandler) fallback(
	ctx context.Context,
	req *ChargeRequest,
	reason string,
) (*ChargeResult, error) {
	// 尝试从缓存读取上次成功的支付结果
	cached, err := h.fallbackCache.Get(ctx, req.OrderID)
	if err == nil {
		cached.Degraded = true
		cached.Message = "支付服务暂时不可用，显示历史数据"
		return cached, nil
	}
	// 缓存也没有，返回排队中
	return &ChargeResult{
		OrderID:  req.OrderID,
		Status:   "pending",
		Degraded: true,
		Message:  "支付排队中，请稍后重试",
	}, nil
}

# chaos-experiment.yaml
experiment:
  name: "payment-service-failure"
  description: "验证支付服务单实例故障时系统的降级效果"
  
  # 实验假设
  hypothesis: "当支付服务一个实例被杀掉时，整体支付成功率应保持在 99% 以上"

  # 爆炸半径控制
  blast_radius:
    target: "payment-service"
    instances: 1  # 只杀一个实例
    percentage: 25 # 不超过实例总数的 25%
  
  # 健康检查
  steady_state:
    - metric: "payment_success_rate"
      threshold: ">= 0.99"
    - metric: "payment_p99_latency"
      threshold: "<= 500ms"
    - metric: "circuit_breaker_state"
      expected: "closed"  # 熔断器不应该被触发
  
  # 自动终止条件
  abort_conditions:
    - metric: "payment_success_rate"
      threshold: "< 0.95"
      action: "rollback"
    - duration: "5m"
      action: "auto-stop"
  
  # 实验步骤
  steps:
    - name: "baseline"
      action: "collect_metrics"
      duration: "2m"
    
    - name: "inject_failure"
      action: "terminate_instance"
      target: "payment-service-canary"
    
    - name: "observe"
      action: "collect_metrics"
      duration: "5m"
    
    - name: "analyze"
      action: "compare_with_hypothesis"

# Prometheus 指标示例

# 熔断器状态（0=closed, 1=open, 2=half-open）
circuit_breaker_state{service="payment",breaker="default"}

# 限流器通过/拒绝速率
rate(rate_limiter_allowed_total{service="api"}[1m])
rate(rate_limiter_rejected_total{service="api"}[1m])

# 降级触发次数
increase(fallback_invoked_total{service="product"}[5m])

# 舱壁使用率
bulkhead_active_concurrent{service="payment"} / bulkhead_max_concurrent{service="payment"}