熔断与限流:滑动窗口的两种算法

代码:src/rate_limiter/{token_bucket, redis_token_bucket}.{h,cc} + src/circuit_breaker/{circuit_breaker}.{h,cc} 场景:防止下游服务雪崩,客户端限流 + 熔断 + 服务降级

微服务架构里,保护下游比"打到最快"更重要。三个核心组件:

  • 限流(Rate Limiter):单位时间最多处理 N 个请求
  • 熔断(Circuit Breaker):下游错误率超阈值,临时熔断,不再打下游
  • 负载均衡(Load Balancer):把请求分散到多个实例(轮询/一致性哈希)

这篇文聚焦前两者——把面试常考的"你怎么做限流"答得有深度

一、限流:三种算法对比

算法突发均匀实现难度适用
Token Bucket允许简单API 限流(可突发的场景)
Sliding Window严格精确中等严格限流(如秒杀)
Leaky Bucket拒绝严格均匀简单平滑流出(如流量整形)

1.1 Token Bucket(令牌桶)

核心:桶里持续加 token,每个请求消费 1 token,无 token 则拒绝。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
class TokenBucket {
 public:
  TokenBucket(double rate, double capacity)
    : rate_(rate), capacity_(capacity), tokens_(capacity), 
      last_refill_(Clock::now()) {}
  
  bool try_acquire(int n = 1) {
    refill();
    if (tokens_ >= n) {
      tokens_ -= n;
      return true;
    }
    return false;
  }

 private:
  void refill() {
    auto now = Clock::now();
    auto elapsed = duration_cast<ms>(now - last_refill_).count() / 1000.0;
    tokens_ = min(capacity_, tokens_ + elapsed * rate_);
    last_refill_ = now;
  }
  
  double rate_;          // 每秒生成多少 token
  double capacity_;      // 桶容量
  double tokens_;        // 当前 token 数
  TimePoint last_refill_;
};

用法:rate=100, capacity=200——突发 200 个请求(桶满),持续 100 QPS。

关键:refill()惰性的——try_acquire() 时才算"应该加几个 token",不用定时器

1.2 Sliding Window(滑动窗口)

核心:维护一个最近 N 个时间单位的请求计数,用 deque 存时间戳

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
class SlidingWindow {
 public:
  SlidingWindow(int max_requests, chrono::seconds window)
    : max_(max_requests), window_(window) {}
  
  bool try_acquire() {
    auto now = Clock::now();
    lock_guard<mutex> lk(mu_);
    // 弹出过期的时间戳
    while (!q_.empty() && now - q_.front() > window_) {
      q_.pop_front();
    }
    if ((int)q_.size() < max_) {
      q_.push_back(now);
      return true;
    }
    return false;
  }
  
 private:
  int max_;
  chrono::seconds window_;
  deque<TimePoint> q_;
  mutex mu_;
};

优点:严格精确——任意 1s 窗口内不超过 N。

缺点:deque 内存随流量涨。1000 QPS × 1s 窗口 = 1000 项。大流量要换 atomic + 分桶(下面 redis 版用这思路)。

1.3 Redis 分布式版:用 Lua 原子

单进程 Token Bucket 不够——多个 RPC client 进程,各自限流 100 QPS,实际 1000 QPS 打到下游。

正解:共享计数器。Redis 存剩余 token 数,Lua 脚本保证原子:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
-- redis_token_bucket.lua
-- KEYS[1] = "rate:user:123"
-- ARGV[1] = rate (每秒)
-- ARGV[2] = capacity
-- ARGV[3] = now (秒, 浮点)
-- ARGV[4] = n (要拿几个)

local key = KEYS[1]
local rate = tonumber(ARGV[1])
local capacity = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local n = tonumber(ARGV[4])

local data = redis.call('HMGET', key, 'tokens', 'last')
local tokens = tonumber(data[1]) or capacity
local last = tonumber(data[2]) or now

-- 补充 token
local elapsed = math.max(0, now - last)
tokens = math.min(capacity, tokens + elapsed * rate)

if tokens >= n then
  tokens = tokens - n
  redis.call('HMSET', key, 'tokens', tokens, 'last', now)
  redis.call('EXPIRE', key, 60)  -- 60s 不用则清掉
  return 1
else
  redis.call('HMSET', key, 'tokens', tokens, 'last', now)
  redis.call('EXPIRE', key, 60)
  return 0
end

关键:整个 refill + acquire 在 Lua 里——Redis 单线程跑 Lua,天然原子

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
// C++ 客户端
class RedisTokenBucket {
  RedisClient* redis_;
  string key_;
  double rate_, capacity_;
  
  bool try_acquire(int n = 1) {
    auto now = get_time_seconds();
    int ok = redis_->eval(LUA_SCRIPT, {key_}, 
                          rate_, capacity_, now, n);
    return ok == 1;
  }
};

性能:单 Redis 跑 Lua 脚本 ~5w QPS——足够


二、熔断:Circuit Breaker 三状态

2.1 状态机

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
       错误率 < 阈值
   ┌─────────────────────┐
   │                     ▼
[CLOSED]              [HALF_OPEN]
   │                     │
   │ 错误率 >= 阈值       │ N 次成功 → CLOSED
   │ (如 50%)           │ 任一失败 → OPEN
   ▼                     │
[OPEN] ─── 30s timeout ──┘
   │ 状态变化由后台线程监控

2.2 实现

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
class CircuitBreaker {
 public:
  CircuitBreaker(int window_size, double error_threshold, 
                 chrono::seconds open_duration)
    : window_(window_size), threshold_(error_threshold), 
      open_duration_(open_duration) {}
  
  bool allow_request() {
    State s = state_.load();
    if (s == State::CLOSED) return true;
    if (s == State::OPEN) {
      // 30s 后进 HALF_OPEN
      if (Clock::now() - opened_at_ > open_duration_) {
        state_.compare_exchange_strong(s, State::HALF_OPEN);
        return true;  // 放一个过去试
      }
      return false;  // 熔断中
    }
    // HALF_OPEN:放一半过去
    if (s == State::HALF_OPEN) {
      if (half_open_inflight_ < max_half_open_) {
        half_open_inflight_++;
        return true;
      }
      return false;
    }
    return false;
  }
  
  void record_success() {
    success_count_++;
    if (state_.load() == State::HALF_OPEN) {
      half_open_success_++;
      if (half_open_success_ >= max_half_open_) {
        state_.store(State::CLOSED);  // 全成功,关掉
        reset();
      }
    }
  }
  
  void record_failure() {
    failure_count_++;
    if (state_.load() == State::HALF_OPEN) {
      state_.store(State::OPEN);  // 还不行,继续开
      opened_at_ = Clock::now();
      return;
    }
    if (state_.load() == State::CLOSED) {
      // 滑动窗口:最近 N 次中,错误率超阈值 → OPEN
      if (total_count_ >= window_ && 
          failure_count_ / total_count_ > threshold_) {
        state_.store(State::OPEN);
        opened_at_ = Clock::now();
      }
    }
  }
  
 private:
  void reset() {
    success_count_ = failure_count_ = total_count_ = 0;
  }
  
  enum class State { CLOSED, OPEN, HALF_OPEN };
  atomic<State> state_{State::CLOSED};
  TimePoint opened_at_;
  int window_;
  double threshold_;
  chrono::seconds open_duration_;
  atomic<int> half_open_inflight_{0};
  atomic<int> half_open_success_{0};
  static constexpr int max_half_open_ = 5;
  
  atomic<int> success_count_{0};
  atomic<int> failure_count_{0};
  atomic<int> total_count_{0};
};

2.3 滑动窗口怎么算错误率

我用的是计数式滑动窗口(上面代码)——success_count + failure_count 达到 window_ 时算一次,然后 reset

生产更精细时间窗口——1s 桶切 10 个 100ms 桶,当前时间往前 1s 的桶内统计:

1
2
3
4
5
6
7
struct Bucket {
  atomic<int> success{0};
  atomic<int> failure{0};
  TimePoint start;  // 100ms 窗口起点
};

vector<Bucket> buckets_(10);  // 1s 内分 10 桶

每次记录时,index = (now / 100ms) % 10,用 CAS 跳过过期的桶:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
void record(bool success) {
  auto now = Clock::now();
  auto bucket_idx = (now.time_since_epoch() / 100ms) % 10;
  auto& b = buckets_[bucket_idx];
  if (now - b.start > 1s) {
    // 桶过期,重置
    b.start = now;
    b.success = b.failure = 0;
  }
  if (success) b.success++;
  else b.failure++;
}

比计数式更"实时"——错误率反应在最近 1s 内的真实分布,不会"100 个老请求 + 10 个新失败 = 9% 错误率,熔断不触发"。


三、整合:RpcClient 怎么用

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
class RpcClient {
  TokenBucket rate_limiter_;     // 本地限流
  CircuitBreaker breaker_;       // 下游熔断
  ServiceDiscovery discovery_;   // 服务发现
  
  RpcResponse call(string service, string method, RpcRequest req) {
    // 1. 限流
    if (!rate_limiter_.try_acquire()) {
      throw RpcException("rate limited");
    }
    
    // 2. 熔断检查
    if (!breaker_.allow_request()) {
      throw RpcException("circuit open");
    }
    
    // 3. 选实例 + 发送
    auto instances = discovery_.discover(service);
    auto* inst = load_balancer_.pick(instances);
    try {
      auto resp = do_call(inst, method, req);
      breaker_.record_success();
      return resp;
    } catch (...) {
      breaker_.record_failure();
      throw;
    }
  }
};

调用顺序:

  1. 限流先 check(快速失败,不浪费下游)
  2. 熔断 check(失败也快速,不到下游)
  3. 选实例 + 真打

关键:限流和熔断分开——一个防"自己打太多",一个防"下游已经坏了"。


四、3 个常被问的难题

4.1 “限流算法选哪个?”

场景推荐
普通 APIToken Bucket(允许突发,实现简单)
秒杀(严格 100 QPS)Sliding Window(精确)
流量整形(平滑输出)Leaky Bucket
分布式(多进程)Redis + Token Bucket Lua

面试回答模板:“默认 Token Bucket,秒杀换 Sliding Window,分布式加 Redis”——一句话答完,深挖再展开。

4.2 “熔断和降级的区别”

概念含义
熔断下游错了,主动断开,暂时不调
降级下游错了,返回兜底数据(默认值、缓存)
限流请求太多了,主动拒绝

顺序:熔断 → 降级 → 限流(从下游到上游,层层保护)。

降级举例:用户服务挂了,订单服务调过去,返回**“用户信息暂时不可用”**(默认值),而不是抛 500。

4.3 “怎么避免’雪崩’?”

雪崩:服务 A 挂 → 调 A 的 B 全排队 → B 的线程池占满 → B 挂 → 调 B 的 C 也挂……

防护三件套:

  1. 超时(B 调 A 设 1s 超时,不要无限等)
  2. 熔断(A 错误率高,B 不再调 A)
  3. 隔离(B 调多个下游,线程池分开,A 挂了不影响 B 调 D)

我项目里:RpcClient 给每个下游一个独立 EventLoopThreadPool——A 挂了只占 A 的线程,不影响 B。


五、未做完的事(留作下篇)

  • 自适应限流:根据下游响应时间动态调限流阈值(类似 TCP BBR)
  • 熔断告警:从 OPEN → HALF_OPEN 转换时,推送到 Prometheus
  • 分布式熔断:用 etcd 共享熔断状态,多 client 协同

这些是工程级的事,生产里有 80% 的价值——但单测 demo 跑不出来,需要真实流量。


相关阅读: