微服务限流熔断降级策略：从雪崩防护到精细化流量治理的实战指南

用户请求 → API Gateway → 订单服务 → 用户服务（鉴权）
                                  → 库存服务（扣减）
                                  → 优惠券服务
                                  → 支付服务 → 第三方支付网关

package ratelimit

import (
	"context"
	"sync"
	"time"
)

// TokenBucket 令牌桶限流器
type TokenBucket struct {
	mu         sync.Mutex
	rate       float64   // 令牌发放速率（个/秒）
	burst      float64   // 桶容量（最大令牌数）
	tokens     float64   // 当前令牌数
	lastUpdate time.Time // 上次补充令牌的时间
}

// NewTokenBucket 创建限流器
// rate: 每秒发放多少令牌
// burst: 桶最多存多少令牌（允许的突发量）
func NewTokenBucket(rate, burst float64) *TokenBucket {
	return &TokenBucket{
		rate:       rate,
		burst:      burst,
		tokens:     burst, // 初始满桶
		lastUpdate: time.Now(),
	}
}

// Allow 尝试获取1个令牌，非阻塞
func (tb *TokenBucket) Allow() bool {
	return tb.AllowN(1)
}

// AllowN 尝试获取n个令牌，非阻塞
func (tb *TokenBucket) AllowN(n float64) bool {
	tb.mu.Lock()
	defer tb.mu.Unlock()

	now := time.Now()
	// 计算距上次过了多久，补充令牌
	elapsed := now.Sub(tb.lastUpdate).Seconds()
	tb.tokens += elapsed * tb.rate
	// 令牌不能超过桶容量
	if tb.tokens > tb.burst {
		tb.tokens = tb.burst
	}
	tb.lastUpdate = now

	// 令牌够不够
	if tb.tokens >= n {
		tb.tokens -= n
		return true
	}
	return false
}

// Wait 阻塞等待直到获取令牌或超时
func (tb *TokenBucket) Wait(ctx context.Context) error {
	for {
		if tb.Allow() {
			return nil
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(10 * time.Millisecond):
			// 10ms 重试一次，避免忙等
		}
	}
}

package ratelimit

import (
	"context"
	"fmt"
	"time"

	"github.com/redis/go-redis/v9"
)

// RedisRateLimiter 基于 Redis 的分布式限流器
// 使用滑动窗口算法
type RedisRateLimiter struct {
	client *redis.Client
	key    string        // 限流 key，如 "rate_limit:order_service"
	rate   int           // 窗口内允许的最大请求数
	window time.Duration // 窗口大小
}

func NewRedisRateLimiter(client *redis.Client, key string, rate int, window time.Duration) *RedisRateLimiter {
	return &RedisRateLimiter{
		client: client,
		key:    key,
		rate:   rate,
		window: window,
	}
}

// Allow 使用 Redis Lua 脚本保证原子性
func (r *RedisRateLimiter) Allow(ctx context.Context) (bool, error) {
	// Lua 脚本：原子性地移除窗口外数据 + 计数 + 判断
	script := `
		local key = KEYS[1]
		local now = tonumber(ARGV[1])
		local window = tonumber(ARGV[2])
		local rate = tonumber(ARGV[3])

		-- 移除窗口外的旧记录
		redis.call('ZREMRANGEBYSCORE', key, 0, now - window)

		-- 获取当前窗口内请求数
		local count = redis.call('ZCARD', key)

		if count < rate then
			-- 未超限，记录当前请求
			redis.call('ZADD', key, now, now .. ':' .. math.random())
			redis.call('PEXPIRE', key, window)
			return 1
		else
			return 0
		end
	`

	now := time.Now().UnixMilli()
	result, err := r.client.Eval(ctx, script, []string{r.key},
		now, r.window.Milliseconds(), r.rate).Int()
	if err != nil {
		return false, fmt.Errorf("redis eval failed: %w", err)
	}
	return result == 1, nil
}

// Sentinel 限流规则配置
FlowRule rule = new FlowRule();
rule.setResource("orderService.createOrder");
rule.setGrade(RuleConstant.FLOW_GRADE_QPS);  // QPS 模式
rule.setCount(1000);                           // 每秒最多 1000 次
rule.setLimitApp("default");                    // 对所有来源生效

// 热点参数限流：按用户ID限流，单用户每秒最多10次
ParamFlowRule paramRule = new ParamFlowRule("orderService.createOrder")
    .setParamIdx(0)        // 第0个参数作为限流维度
    .setGrade(RuleConstant.FLOW_GRADE_QPS)
    .setCount(10);          // 单用户 QPS 上限
// 特殊用户（VIP）放宽到 100
paramRule.setParamItem(new ParamFlowItem()
    .setObject("user_vip_123")
    .setClassType(String.class.getName())
    .setCount(100));

// 加载规则
FlowRuleManager.loadRules(Collections.singletonList(rule));
ParamFlowRuleManager.loadRules(Collections.singletonList(paramRule));

// 业务代码中使用
@SentinelResource(value = "orderService.createOrder",
    blockHandler = "createOrderBlockHandler")
public Order createOrder(String userId, OrderRequest req) {
    // 正常业务逻辑
    return orderRepository.save(req);
}

// 限流后的降级处理
public Order createOrderBlockHandler(String userId, OrderRequest req,
    BlockException ex) {
    // 返回默认订单或抛出友好提示
    throw new ServiceException("系统繁忙，请稍后重试", 429);
}

              失败率超过阈值
  ┌─────────┐ ──────────────→ ┌─────────┐
  │ CLOSED  │                │  OPEN   │
  │ (正常)   │ ←───────────── │ (熔断)   │
  └─────────┘  探测成功        └─────────┘
       ↑                        │
       │                        │ 等待冷却时间结束
       │                        ↓
       │                   ┌──────────┐
       └────────────────── │HALF_OPEN │
           探测成功          │ (半开)    │
                           └──────────┘
                              │ 探测失败
                              ↓
                           ┌─────────┐
                           │  OPEN   │
                           └─────────┘

<!-- pom.xml 只引入需要的模块 -->
<dependencies>
    <dependency>
        <groupId>io.github.resilience4j</groupId>
        <artifactId>resilience4j-circuitbreaker</artifactId>
        <version>2.2.0</version>
    </dependency>
    <dependency>
        <groupId>io.github.resilience4j</groupId>
        <artifactId>resilience4j-ratelimiter</artifactId>
        <version>2.2.0</version>
    </dependency>
    <dependency>
        <groupId>io.github.resilience4j</groupId>
        <artifactId>resilience4j-bulkhead</artifactId>
        <version>2.2.0</version>
    </dependency>
</dependencies>

# application.yml
resilience4j:
  circuitbreaker:
    configs:
      default:
        # 滑动窗口配置
        slidingWindowType: COUNT_BASED       # 基于请求数统计
        slidingWindowSize: 100               # 最近 100 次请求
        minimumNumberOfCalls: 20            # 至少 20 次请求才开始计算
        # 失败判定
        failureRateThreshold: 50             # 失败率 >= 50% 触发熔断
        slowCallRateThreshold: 60            # 慢调用比例 >= 60% 触发熔断
        slowCallDurationThreshold: 2s        # 超过 2 秒算慢调用
        # 状态转换
        waitDurationInOpenState: 30s         # 熔断后等 30 秒再半开
        permittedNumberOfCallsInHalfOpenState: 10  # 半开状态放行 10 个探测请求
        # 哪些异常算失败
        recordExceptions:
          - java.io.IOException
          - java.util.concurrent.TimeoutException
          - feign.FeignException.GatewayTimeout
        ignoreExceptions:
          - com.example.BusinessException    # 业务异常不计入熔断
    instances:
      paymentService:
        baseConfig: default
        # 支付服务单独调低阈值，因为支付链路更敏感
        failureRateThreshold: 30
        slowCallDurationThreshold: 1s
      userService:
        baseConfig: default
        # 用户服务允许更高的失败率，因为可以做降级
        failureRateThreshold: 60

package circuitbreaker

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

// State 熔断器状态
type State int

const (
	StateClosed   State = iota // 正常
	StateOpen                  // 熔断
	StateHalfOpen              // 半开
)

// Settings 熔断器配置
type Settings struct {
	Name          string
	MaxRequests   uint32        // 半开状态允许的探测请求数
	Interval      time.Duration // Closed 状态下的统计周期
	Timeout       time.Duration // Open 状态的冷却时间
	ReadyToTrip   func(counts Counts) bool // 判断是否触发熔断
	OnStateChange func(name string, from, to State)
}

// Counts 请求计数
type Counts struct {
	Requests            uint32
	TotalSuccesses      uint32
	TotalFailures       uint32
	ConsecutiveFailures uint32
}

// CircuitBreaker 熔断器
type CircuitBreaker struct {
	mu          sync.Mutex
	settings    Settings
	state       State
	generation  uint64
	expiry      time.Time
	counts      Counts
}

// NewCircuitBreaker 创建熔断器
func NewCircuitBreaker(s Settings) *CircuitBreaker {
	if s.ReadyToTrip == nil {
		// 默认策略：连续失败 5 次触发熔断
		s.ReadyToTrip = func(c Counts) bool {
			return c.ConsecutiveFailures >= 5
		}
	}
	cb := &CircuitBreaker{settings: s}
	cb.toNewGeneration(time.Now())
	return cb
}

// Execute 执行请求，自动处理熔断逻辑
func (cb *CircuitBreaker) Execute(req func() error) error {
	generation, err := cb.beforeRequest()
	if err != nil {
		return err
	}

	err = req()

	cb.afterRequest(generation, err)
	return err
}

func (cb *CircuitBreaker) beforeRequest() (uint64, error) {
	cb.mu.Lock()
	defer cb.mu.Unlock()

	now := time.Now()
	state, generation := cb.currentState(now)

	if state == StateOpen {
		// 熔断中，直接返回错误，不执行请求
		return generation, ErrCircuitOpen
	}
	return generation, nil
}

func (cb *CircuitBreaker) afterRequest(before uint64, err error) {
	cb.mu.Lock()
	defer cb.mu.Unlock()

	now := time.Now()
	state, generation := cb.currentState(now)
	if before != generation {
		// 状态已经变了，丢弃这次计数
		return
	}

	cb.counts.Requests++
	if err != nil {
		cb.counts.TotalFailures++
		cb.counts.ConsecutiveFailures++
	} else {
		cb.counts.TotalSuccesses++
		cb.counts.ConsecutiveFailures = 0
	}

	// 检查是否需要切换状态
	if cb.state == StateClosed && cb.settings.ReadyToTrip(cb.counts) {
		cb.setState(StateOpen, now)
	} else if cb.state == StateHalfOpen && cb.counts.Requests >= cb.settings.MaxRequests {
		// 半开状态下探测请求已用完，根据结果决定状态
		if cb.counts.TotalSuccesses > 0 {
			cb.setState(StateClosed, now)
		} else {
			cb.setState(StateOpen, now)
		}
	}
}

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

func (cb *CircuitBreaker) currentState(now time.Time) (State, uint64) {
	switch cb.state {
	case StateClosed:
		if !cb.expiry.IsZero() && now.After(cb.expiry) {
			cb.toNewGeneration(now)
		}
	case StateOpen:
		if now.After(cb.expiry) {
			// 冷却结束，进入半开
			cb.setState(StateHalfOpen, now)
		}
	}
	return cb.state, cb.generation
}

func (cb *CircuitBreaker) setState(state State, now time.Time) {
	if cb.state == state {
		return
	}
	prev := cb.state
	cb.state = state
	cb.toNewGeneration(now)
	if cb.settings.OnStateChange != nil {
		cb.settings.OnStateChange(cb.settings.Name, prev, state)
	}
}

func (cb *CircuitBreaker) toNewGeneration(now time.Time) {
	cb.generation++
	cb.counts = Counts{}

	switch cb.state {
	case StateClosed:
		if cb.settings.Interval == 0 {
			cb.expiry = time.Time{} // 不过期
		} else {
			cb.expiry = now.Add(cb.settings.Interval)
		}
	case StateOpen:
		cb.expiry = now.Add(cb.settings.Timeout)
	case StateHalfOpen:
		cb.expiry = time.Time{} // 不过期，等探测请求用完
	}
}

cb := circuitbreaker.NewCircuitBreaker(circuitbreaker.Settings{
    Name:        "payment-service",
    MaxRequests: 5,                    // 半开时放行 5 个探测
    Interval:    60 * time.Second,     // Closed 统计周期 60 秒
    Timeout:     30 * time.Second,     // Open 冷却 30 秒
    ReadyToTrip: func(c circuitbreaker.Counts) bool {
        // 失败率超过 50% 且至少 10 次请求才触发
        if c.Requests < 10 {
            return false
        }
        failureRate := float64(c.TotalFailures) / float64(c.Requests)
        return failureRate >= 0.5
    },
    OnStateChange: func(name string, from, to circuitbreaker.State) {
        log.Printf("[CircuitBreaker] %s: %v → %v", name, from, to)
        // 状态变化时推送告警
        alert.Send(fmt.Sprintf("熔断器 %s 从 %v 切换到 %v", name, from, to))
    },
})

// 在业务代码中使用
err := cb.Execute(func() error {
    return paymentClient.Charge(ctx, order)
})
if errors.Is(err, circuitbreaker.ErrCircuitOpen) {
    // 熔断已打开，执行降级逻辑
    return fallbackPayment(ctx, order)
}

# Istio DestinationRule 配置 Envoy 熔断
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: payment-service-cb
spec:
  host: payment-service.default.svc.cluster.local
  trafficPolicy:
    outlierDetection:
      # 连续 5xx 错误数达到 5 触发驱逐
      consecutive5xxErrors: 5
      # 连续网关错误（502/503/504）达到 3 触发驱逐
      consecutiveGatewayErrors: 3
      # 驱逐时间间隔（最少驱逐多久）
      interval: 30s
      # 驱逐时长
      baseEjectionTime: 30s
      # 最大驱逐比例：最多驱逐 50% 的实例
      maxEjectionPercent: 50
      # 最小健康比例：至少保持 50% 实例可用
      minHealthPercent: 50
    connectionPool:
      tcp:
        maxConnections: 100        # 最大连接数
        connectTimeout: 2s         # 连接超时
      http:
        http1MaxPendingRequests: 50    # 最大等待中的请求
        http2MaxRequests: 200          # 最大并发请求
        maxRequestsPerConnection: 10   # 每个连接最大请求数
        maxRetries: 2                  # 最大重试次数

// PaymentService 支付服务，包含降级链
type PaymentService struct {
	primaryPayment  PaymentClient   // 主支付通道
	fallbackPayment PaymentClient   // 备用支付通道
	cache           CacheClient     // 缓存
	cb              *circuitbreaker.CircuitBreaker // 熔断器
}

// Charge 支付，包含三级降级策略
func (s *PaymentService) Charge(ctx context.Context, order *Order) (*PaymentResult, error) {
	// 策略1：主通道正常 → 正常支付
	result, err := s.cb.Execute(func() error {
		result, err := s.primaryPayment.Charge(ctx, order)
		if err == nil {
			order.Result = result
		}
		return err
	})

	if err == nil {
		return result, nil
	}

	// 策略2：主通道熔断 → 切换备用通道
	if errors.Is(err, circuitbreaker.ErrCircuitOpen) {
		log.Printf("primary payment circuit open, switching to fallback")
		result, err := s.fallbackPayment.Charge(ctx, order)
		if err == nil {
			return result, nil
		}
	}

	// 策略3：两个通道都不行 → 异步处理
	// 先返回"处理中"，把请求丢到队列，后台重试
	if order.Amount < 10000 { // 小额订单可以异步
		if err := s.enqueueAsyncPayment(order); err != nil {
			return nil, fmt.Errorf("payment unavailable, please retry later")
		}
		return &PaymentResult{
			Status:  "processing",
			Message: "支付处理中，请稍后查看结果",
		}, nil
	}

	// 大额订单不允许异步，直接拒绝
	return nil, fmt.Errorf("payment service unavailable")
}

// 从配置中心读取降级配置
type DegradeConfig struct {
	// 功能开关
	EnableRecommendation bool `json:"enable_recommendation"` // 推荐功能
	EnableComments       bool `json:"enable_comments"`       // 评论功能
	EnableCoupons        bool `json:"enable_coupons"`        // 优惠券功能
	// 降级策略
	RecommendationDegrade string `json:"recommendation_degrade"` // "default" | "cache" | "disabled"
	// 限流阈值（可动态调整）
	OrderRateLimit int `json:"order_rate_limit"` // 下单 QPS 限制
}

// 降级管理器
type DegradeManager struct {
	config *DegradeConfig
	mu     sync.RWMutex
}

// CheckFeature 检查功能是否可用
func (dm *DegradeManager) CheckFeature(feature string) bool {
	dm.mu.RLock()
	defer dm.mu.RUnlock()
	switch feature {
	case "recommendation":
		return dm.config.EnableRecommendation
	case "comments":
		return dm.config.EnableComments
	case "coupons":
		return dm.config.EnableCoupons
	}
	return true // 默认开启
}

// UpdateConfig 配置中心回调，热更新降级配置
func (dm *DegradeManager) UpdateConfig(newConfig *DegradeConfig) {
	dm.mu.Lock()
	dm.config = newConfig
	dm.mu.Unlock()
	log.Printf("degrade config updated: %+v", newConfig)
}

                        用户请求
                           │
                           ▼
                 ┌──────────────────┐
                 │   API Gateway    │  ← 全局限流（令牌桶）
                 │  (Kong/APISIX)   │     按 API 限流，按租户限流
                 └────────┬─────────┘
                          │
              ┌───────────┼───────────┐
              ▼           ▼           ▼
         ┌────────┐ ┌────────┐ ┌────────┐
         │订单服务 │ │用户服务 │ │搜索服务 │  ← 服务级限流（Sentinel/gobreaker）
         │        │ │        │ │        │     按接口限流，按用户限流
         │熔断器 A │ │熔断器 B │ │熔断器 C │  ← 调用下游时熔断
         └───┬────┘ └───┬────┘ └───┬────┘
             │          │          │
             ▼          ▼          ▼
         ┌────────┐ ┌────────┐ ┌────────┐
         │支付服务 │ │认证服务 │ │数据库   │  ← 资源级保护
         │        │ │        │ │        │     连接池限流，慢SQL熔断
         │降级策略 │ │降级策略 │ │        │
         └────────┘ └────────┘ └────────┘

API Gateway 超时 5s
    └─ 订单服务超时 3s
        └─ 支付服务超时 2s
            └─ 第三方支付超时 1.5s

// 渐进式恢复：在熔断器恢复后控制流量比例
type GradualRecovery struct {
	mu              sync.Mutex
	recoveryPhase   int      // 0=关闭, 1=10%, 2=30%, 3=100%
	phaseStartTime  time.Time
}

func (gr *GradualRecovery) Allow() bool {
	gr.mu.Lock()
	defer gr.mu.Unlock()

	ratios := []float64{0.1, 0.3, 1.0} // 渐进比例
	if gr.recoveryPhase >= len(ratios) {
		return true // 完全恢复
	}

	// 检查是否该进入下一阶段
	if time.Since(gr.phaseStartTime) > 30*time.Second {
		gr.recoveryPhase++
		gr.phaseStartTime = time.Now()
	}

	// 按比例放行
	ratio := ratios[gr.recoveryPhase]
	if rand.Float64() < ratio {
		return true
	}
	return false
}

# Prometheus 抓取熔断器指标（Resilience4j 暴露的 metrics）
scrape_configs:
  - job_name: 'resilience4j'
    metrics_path: '/actuator/prometheus'
    static_configs:
      - targets: ['order-service:8080', 'payment-service:8080']

# 熔断器状态（0=CLOSED, 1=OPEN, 2=HALF_OPEN）
resilience4j_circuitbreaker_state

# 熔断器调用失败率
resilience4j_circuitbreaker_failure_rate

# 被限流的请求速率
rate(resilience4j_ratelimiter_available_permissions[1m])

# 告警规则：熔断器打开超过1分钟
ALERT CircuitBreakerOpen
IF resilience4j_circuitbreaker_state == 1
FOR 1m
LABELS { severity = "critical" }
ANNOTATIONS {
  summary = "熔断器 {{ $labels.name }} 已打开",
  description = "熔断器已打开超过1分钟，请检查下游服务状态"
}

安全 QPS = 基准压测拐点 QPS × 安全系数(0.6-0.7)
限流阈值 = 安全 QPS × 实例数 × 负载不均衡系数(0.8)