📖 第09课：限流与熔断

阶段二：RESTful进阶

限流(Rate Limiting)保护API不被滥用，熔断(Circuit Breaker)防止故障级联。两者是API可靠性的核心防线——没有限流，一个恶意客户端就能拖垮整个服务；没有熔断，一个下游故障就能引发雪崩效应。

🚦 限流算法详解

1. 固定窗口计数器

// 固定窗口：每分钟一个计数器，简单但有边界突刺问题
class FixedWindowLimiter {
  constructor(limit, windowMs) {
    this.limit = limit;       // 窗口内最大请求数
    this.windowMs = windowMs; // 窗口大小（毫秒）
    this.counters = new Map(); // key → {count, windowStart}
  }

  isAllowed(key) {
    const now = Date.now();
    const windowStart = Math.floor(now / this.windowMs) * this.windowMs;

    let record = this.counters.get(key);
    if (!record || record.windowStart !== windowStart) {
      record = { count: 0, windowStart };
      this.counters.set(key, record);
    }

    record.count++;
    const remaining = Math.max(0, this.limit - record.count);
    const resetAt = windowStart + this.windowMs;

    return {
      allowed: record.count <= this.limit,
      limit: this.limit,
      remaining,
      resetAt,
      retryAfter: record.count > this.limit ? Math.ceil((resetAt - now) / 1000) : 0,
    };
  }
}

// 问题：窗口边界突刺
// [00:00-01:00] 最后1秒涌入100请求 + [01:00-02:00] 第1秒涌入100请求
// 实际2秒内处理了200请求，远超预期

2. 滑动窗口日志

// 滑动窗口：记录每个请求的时间戳，精确但内存开销大
class SlidingWindowLogLimiter {
  constructor(limit, windowMs) {
    this.limit = limit;
    this.windowMs = windowMs;
    this.logs = new Map(); // key → [timestamp, ...]
  }

  isAllowed(key) {
    const now = Date.now();
    const windowStart = now - this.windowMs;

    if (!this.logs.has(key)) this.logs.set(key, []);

    // 清理过期记录
    const logs = this.logs.get(key).filter(ts => ts > windowStart);
    this.logs.set(key, logs);

    const allowed = logs.length < this.limit;
    if (allowed) logs.push(now);

    return {
      allowed,
      limit: this.limit,
      remaining: Math.max(0, this.limit - logs.length - (allowed ? 0 : 0)),
      resetAt: logs.length > 0 ? logs[0] + this.windowMs : now + this.windowMs,
    };
  }
}

3. 令牌桶（Token Bucket）—— 最推荐

// 令牌桶：匀速生成令牌，允许突发流量，最灵活
class TokenBucketLimiter {
  constructor(capacity, refillRate) {
    this.capacity = capacity;   // 桶容量（最大令牌数=允许突发量）
    this.refillRate = refillRate; // 每秒填充令牌数
    this.buckets = new Map();    // key → {tokens, lastRefill}
  }

  isAllowed(key, tokensNeeded = 1) {
    const now = Date.now();
    let bucket = this.buckets.get(key);

    if (!bucket) {
      bucket = { tokens: this.capacity, lastRefill: now };
      this.buckets.set(key, bucket);
    }

    // 计算应填充的令牌数
    const elapsed = (now - bucket.lastRefill) / 1000;
    const tokensToAdd = Math.floor(elapsed * this.refillRate);
    bucket.tokens = Math.min(this.capacity, bucket.tokens + tokensToAdd);
    bucket.lastRefill = now;

    // 尝试消费令牌
    const allowed = bucket.tokens >= tokensNeeded;
    if (allowed) bucket.tokens -= tokensNeeded;

    return {
      allowed,
      limit: this.capacity,
      remaining: Math.floor(bucket.tokens),
      retryAfter: allowed ? 0 : Math.ceil((tokensNeeded - bucket.tokens) / this.refillRate),
    };
  }
}

4. 漏桶（Leaky Bucket）

// 漏桶：请求以恒定速率处理，适合消息队列
class LeakyBucketLimiter {
  constructor(capacity, leakRate) {
    this.capacity = capacity;    // 桶容量
    this.leakRate = leakRate;    // 每秒漏出（处理）请求数
    this.buckets = new Map();
  }

  isAllowed(key) {
    const now = Date.now();
    let bucket = this.buckets.get(key);

    if (!bucket) {
      bucket = { queue: 0, lastLeak: now };
      this.buckets.set(key, bucket);
    }

    // 计算漏出的请求数
    const elapsed = (now - bucket.lastLeak) / 1000;
    const leaked = Math.floor(elapsed * this.leakRate);
    bucket.queue = Math.max(0, bucket.queue - leaked);
    bucket.lastLeak = now;

    // 尝试加入队列
    const allowed = bucket.queue < this.capacity;
    if (allowed) bucket.queue++;

    return {
      allowed,
      limit: this.capacity,
      remaining: Math.max(0, this.capacity - bucket.queue),
      retryAfter: allowed ? 0 : Math.ceil((bucket.queue - this.capacity + 1) / this.leakRate),
    };
  }
}

四种算法对比

算法	突发支持	内存	精度	推荐场景
固定窗口	❌	低	低	简单限流、不严格
滑动窗口日志	✅	高	高	精确限流
令牌桶	✅	低	高	✅ 通用推荐
漏桶	❌	低	高	恒定速率处理

🔌 熔断器实现

// circuit-breaker.js - 熔断器
class CircuitBreaker {
  constructor(options = {}) {
    this.failureThreshold = options.failureThreshold || 5;    // 失败次数阈值
    this.successThreshold = options.successThreshold || 3;    // 恢复成功次数阈值
    this.timeout = options.timeout || 30000;                   // 熔断等待时间(ms)
    this.monitorInterval = options.monitorInterval || 10000;   // 监控间隔

    this.state = 'CLOSED';   // CLOSED, OPEN, HALF_OPEN
    this.failureCount = 0;
    this.successCount = 0;
    this.lastFailureTime = null;
    this.listeners = [];
  }

  // 执行受保护的函数
  async execute(fn) {
    if (this.state === 'OPEN') {
      if (Date.now() - this.lastFailureTime >= this.timeout) {
        this.state = 'HALF_OPEN';
        this.successCount = 0;
        this.notify('HALF_OPEN');
      } else {
        const remaining = Math.ceil((this.timeout - (Date.now() - this.lastFailureTime)) / 1000);
        throw new Error(`熔断器开启，请 ${remaining} 秒后重试`);
      }
    }

    try {
      const result = await fn();
      this.onSuccess();
      return result;
    } catch (error) {
      this.onFailure();
      throw error;
    }
  }

  onSuccess() {
    this.failureCount = 0;

    if (this.state === 'HALF_OPEN') {
      this.successCount++;
      if (this.successCount >= this.successThreshold) {
        this.state = 'CLOSED';
        this.successCount = 0;
        this.notify('CLOSED');
      }
    }
  }

  onFailure() {
    this.failureCount++;
    this.lastFailureTime = Date.now();

    if (this.state === 'HALF_OPEN') {
      this.state = 'OPEN';
      this.notify('OPEN');
    } else if (this.failureCount >= this.failureThreshold) {
      this.state = 'OPEN';
      this.notify('OPEN');
    }
  }

  // 获取当前状态
  getState() {
    return {
      state: this.state,
      failureCount: this.failureCount,
      successCount: this.successCount,
      lastFailureTime: this.lastFailureTime,
      config: {
        failureThreshold: this.failureThreshold,
        successThreshold: this.successThreshold,
        timeout: this.timeout,
      },
    };
  }

  // 重置
  reset() {
    this.state = 'CLOSED';
    this.failureCount = 0;
    this.successCount = 0;
    this.lastFailureTime = null;
  }

  // 事件监听
  onStateChange(listener) {
    this.listeners.push(listener);
  }

  notify(newState) {
    this.listeners.forEach(fn => fn(newState, this.getState()));
  }
}

module.exports = CircuitBreaker;

🛠️ 实战：限流+熔断完整API

// protected-api.js
const express = require('express');
const { TokenBucketLimiter } = require('./rate-limiters');
const CircuitBreaker = require('./circuit-breaker');

const app = express();
app.use(express.json());

// ===== 限流配置 =====

// 全局限流：每秒10个请求，突发20
const globalLimiter = new TokenBucketLimiter(20, 10);

// 用户限流：每秒5个请求，突发10
const userLimiters = new Map();
function getUserLimiter(userId) {
  if (!userLimiters.has(userId)) {
    userLimiters.set(userId, new TokenBucketLimiter(10, 5));
  }
  return userLimiters.get(userId);
}

// API限流中间件
function rateLimit(options = {}) {
  return (req, res, next) => {
    // 全局限流
    const globalResult = globalLimiter.isAllowed('global');
    if (!globalResult.allowed) {
      res.setHeader('Retry-After', globalResult.retryAfter);
      return res.status(429).json({
        success: false,
        error: { code: 'RATE_LIMIT_GLOBAL', message: '服务器繁忙，请稍后重试' },
      });
    }

    // 用户限流
    const userId = req.headers['x-user-id'] || req.ip;
    const userLimiter = getUserLimiter(userId);
    const userResult = userLimiter.isAllowed(userId);

    // 设置限流响应头（无论是否超限都设置）
    res.setHeader('X-RateLimit-Limit', userResult.limit.toString());
    res.setHeader('X-RateLimit-Remaining', userResult.remaining.toString());
    res.setHeader('X-RateLimit-Reset', new Date(userResult.resetAt).toISOString());

    if (!userResult.allowed) {
      res.setHeader('Retry-After', userResult.retryAfter.toString());
      return res.status(429).json({
        success: false,
        error: {
          code: 'RATE_LIMIT_USER',
          message: `请求频率超限，请 ${userResult.retryAfter} 秒后重试`,
        },
      });
    }

    next();
  };
}

// ===== 熔断器配置 =====

const dbBreaker = new CircuitBreaker({
  failureThreshold: 3,
  successThreshold: 2,
  timeout: 10000,
});

dbBreaker.onStateChange((state, info) => {
  console.log(`🔴 熔断器状态变更: ${state}`, info);
});

// 模拟不稳定的数据库
let dbCallCount = 0;
function unstableDB() {
  return new Promise((resolve, reject) => {
    dbCallCount++;
    // 每第5次调用失败
    if (dbCallCount % 5 === 0) {
      reject(new Error('数据库连接超时'));
    } else {
      resolve({ id: 1, name: '测试数据', value: Math.random() });
    }
  });
}

// ===== 路由 =====

// 受限流保护的端点
app.get('/api/data', rateLimit(), async (req, res) => {
  try {
    const data = await dbBreaker.execute(() => unstableDB());
    res.json({ success: true, data, breaker: dbBreaker.getState() });
  } catch (err) {
    const breakerState = dbBreaker.getState();
    if (breakerState.state === 'OPEN') {
      res.status(503).json({
        success: false,
        error: { code: 'CIRCUIT_OPEN', message: '服务暂时不可用，熔断器已开启' },
        meta: { breakerState, retryAfter: Math.ceil(breakerState.config.timeout / 1000) },
      });
    } else {
      res.status(500).json({
        success: false,
        error: { code: 'UPSTREAM_ERROR', message: err.message },
        meta: { breakerState },
      });
    }
  }
});

// 查看熔断器状态
app.get('/api/breaker/status', (req, res) => {
  res.json({ success: true, data: dbBreaker.getState() });
});

// 手动重置熔断器
app.post('/api/breaker/reset', (req, res) => {
  dbBreaker.reset();
  res.json({ success: true, data: { message: '熔断器已重置' } });
});

app.listen(3000, () => console.log('🚦 限流+熔断API运行在 http://localhost:3000'));

测试限流和熔断

# 测试1: 正常请求
curl -s http://localhost:3000/api/data -H "X-User-ID: user1" | jq .

# 测试2: 限流响应头
curl -I http://localhost:3000/api/data -H "X-User-ID: user1"
# 观察 X-RateLimit-Limit, X-RateLimit-Remaining

X-RateLimit-Limit: 10
X-RateLimit-Remaining: 9
X-RateLimit-Reset: 2025-01-15T10:01:00.000Z

# 测试3: 触发限流（快速发送20+请求）
for i in $(seq 1 20); do
  curl -s http://localhost:3000/api/data -H "X-User-ID: user1" | jq -c '.success // .error.code'
done

# 测试4: 查看熔断器状态
curl -s http://localhost:3000/api/breaker/status | jq .

# 测试5: 触发熔断（连续失败后）
# 重置计数后连续请求
curl -s -X POST http://localhost:3000/api/breaker/reset
for i in $(seq 1 10); do
  curl -s http://localhost:3000/api/data -H "X-User-ID: user1" 2>/dev/null
  sleep 0.1
done
# 查看熔断器是否打开
curl -s http://localhost:3000/api/breaker/status | jq .

📝 本课小结

四种限流算法：固定窗口→滑动窗口→令牌桶→漏桶，令牌桶最推荐
令牌桶允许突发流量，适合API场景；漏桶恒定速率，适合队列场景
限流响应必须包含X-RateLimit-*头和Retry-After
熔断器三种状态：CLOSED(正常)→OPEN(熔断)→HALF_OPEN(试探)
熔断器防止故障级联，是微服务可靠性的关键
全局限流+用户限流双重保护

💪 练习

练习1：实现分层限流

设计三层限流：全局限流(1000/s)、用户限流(100/s)、API端点限流(POST /api/orders: 10/s)。

练习2：分布式限流

使用Redis实现令牌桶限流，支持多实例部署。提示：用Redis的INCR+EXPIRE实现原子计数。

🏆 本课成就：流量守卫

✅ 掌握四种限流算法的原理和实现
✅ 实现令牌桶限流器
✅ 实现熔断器（三态切换）
✅ 组合限流+熔断保护API
✅ 使用标准限流响应头