diff --git a/hzh/AI/harness.pdf b/hzh/AI/harness.pdf new file mode 100644 index 0000000..8a84f8f Binary files /dev/null and b/hzh/AI/harness.pdf differ diff --git a/hzh/GO/协程池.md b/hzh/GO/协程池.md new file mode 100644 index 0000000..f6c13cf --- /dev/null +++ b/hzh/GO/协程池.md @@ -0,0 +1,408 @@ +--- +tags: [go, goroutine, concurrency, pool] +create time: 2026-05-30 14:20 +--- + +# Go 协程池 + +## 概述 + +梳理 Go 中协程池（Goroutine Pool）的设计动机与两种主流实现方案——基于 channel 的轻量级池子和 `golang.ccpool` 风格的成熟方案，分析其核心模式、线程安全保证以及适用边界。 + +## 正文 + +### 一、为什么需要协程池 + +> [!question] 思考：如果一个服务每秒收到 10000 个请求，每个请求启动一个 goroutine，会发生什么？ + +Go 的 goroutine 极其轻量——初始栈仅 2KB，创建成本约几个纳秒——所以 Go 社区流行"别管它，随时起一个"的信条。**但轻量不等于没有成本**。goroutine 仍然占用内存、参与 GC scan、需要调度器时间片。不加控制的 goroutine 爆炸在真实系统中屡见不鲜： + +```go +// ❌ 灾难写法 +for _, item := range items { + go process(item) // 如果 items 有 100 万个元素…… +} +time.Sleep(time.Hour) // 等所有完成——但这也不是可靠的做法 +``` + +当 goroutine 数量暴增时会导致： +- **内存暴涨**：即便每个只有 2KB，百万级就是 GB 级别 +- **GC 压力**：大量短命对象触发频繁 Full GC +- **CPU 抖动**：太多 runnable goroutine 导致调度竞争 + +| 策略 | 控制粒度 | 实现复杂度 | 适用场景 | +|------|---------|-----------|---------| +| 直接并发 | 无 | 最低 | 小批量、短时间任务 | +| Worker 池 | goroutine 数量 | 中等 | 中等并发、需限流 | +| Semaphore (channel) | 最大并发数 | 低 | 快速限速 | + +### 二、Channel + WaitGroup：最简协程池 + +这是最经典的实现方式——用一个固定数量的 worker 集合加一个任务 channel： + +```go +func Worker(id int, jobs <-chan int, results chan<- int, wg *sync.WaitGroup) { + defer wg.Done() + for j := range jobs { // range 自动退出——jobs channel 关闭后结束 + fmt.Printf("Worker %d started job %d\n", id, j) + time.Sleep(time.Second) + results <- j * 2 + } +} + +func main() { + const numWorkers = 3 + jobs := make(chan int, 100) // 带缓冲的任务通道 + results := make(chan int, 100) // 结果通道 + + var wg sync.WaitGroup + for w := 0; w < numWorkers; w++ { + wg.Add(1) + go Worker(w, jobs, results, &wg) + } + + // 投递任务 + for j := 1; j <= 10; j++ { + jobs <- j + } + close(jobs) // 关键：关闭 channel，worker 的 range 才能退出 + + go func() { + wg.Wait() + close(results) + }() + + // 消费结果 + for r := range results { + fmt.Println("result:", r) + } +} +``` + +这个模式的精妙之处在于**用 channel 的语义替代了锁**： +- `range jobs` 天然处理了"取任务 → 执行 → 取下一个"的循环 +- `close(jobs)` 作为停止信号，worker 不用轮询或检查 cancel +- `WaitGroup` 确保所有 worker 完成后再关闭 results + +> [!tip] Channel 缓冲大小的选择 +> +> - 太大：失去限速效果，退化成全并发 +> - 太小：worker 频繁阻塞在取任务上，吞吐量下降 +> - **经验值**：等于或略大于 worker 数量（如 1-3 倍）即可，除非你确信生产者远快于消费者 + +#### 2.1 结构示意 + +```mermaid +flowchart LR + subgraph Producer["生产者"] + P["task 1, 2, 3..."] + end + + subgraph Pool["协程池"] + J["jobs chan\n(buffered)"] + W1["Worker 1"] + W2["Worker 2"] + W3["Worker 3"] + J --> W1 + J --> W2 + J --> W3 + end + + R["results chan"] + W1 --> R + W2 --> R + W3 --> R + + P --> J + R --> Consumer["消费者"] + + style Pool fill:#e8f0fe,stroke:#1a73e8 + style J fill:#fff3e0,stroke:#f9a825 +``` + +### 三、函数式选项：更友好的 API 封装 + +实际项目中通常会把上面的 boilerplate 封装成一个易用的函数： + +```go +// GoPool 是一个简单的协程池 +type GoPool struct { + workers int + queue chan func() +} + +// NewGoPool 创建一个指定 worker 数量的协程池 +func NewGoPool(workers int) *GoPool { + return &GoPool{ + workers: workers, + queue: make(chan func(), 1000), // 队列缓冲 + } +} + +// Start 启动 worker，返回 shutdown 函数 +func (p *GoPool) Start(ctx context.Context) { + var wg sync.WaitGroup + for i := 0; i < p.workers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case fn, ok := <-p.queue: + if !ok { + return // channel 关闭，退出 + } + fn() + case <-ctx.Done(): // 支持外部取消 + return + } + } + }() + } + + // 返回 shutdown 回调：关闭 channel + 等待 worker 退出 + p.shutdown = func() { + close(p.queue) + wg.Wait() + } +} +``` + +使用变得极其简洁： + +```go +pool := NewGoPool(10) +pool.Start(context.Background()) + +// 提交异步任务 +pool.Submit(func() { + doSomething() +}) + +// 清理 +pool.Shutdown() // close(channel) + WaitGroup.Wait() +``` + +> [!note] Submit 之后如何拿到结果？ +> +> 上面的版本是 fire-and-forget 模式（不返回值）。如果需要收集结果，让匿名函数写入一个共享 channel，或者用 `sync.Once` / `errgroup.Group` 做同步等待。这也是为什么下面要引入更成熟的工具。 + +### 四、errgroup：官方级别的简化方案 + +> [!important] 先看这里 +> +> `golang.org/x/sync/errgroup` 虽然不是传统意义上的"池"，但它解决了最常见的并发痛点：**有限度并发 + 错误传播 + 取消传递**。大多数情况下不需要自己写池子。 + +```go +func downloadUrls(urls []string) error { + g, ctx := errgroup.WithContext(context.Background()) + g.SetLimit(10) // ⚡ 限制最多 10 个并发 goroutine + + for _, url := range urls { + u := url // 注意循环变量捕获陷阱 + g.Go(func() error { + select { + case <-ctx.Done(): // 任意一个失败，全部取消 + return ctx.Err() + default: + return fetch(u) + } + }) + } + + return g.Wait() // 等全部完成，或第一个错误返回 +} +``` + +`errgroup` 的核心价值： +- **SetLimit(10)**：一行代码搞定并发限制，本质上是 channel semaphore +- **Go(fn) 返回 error**：任何一个 goroutine panic 或返回非 nil error，整个组取消并提前退出 +- **WithCancel + WaitGroup 一体化**：不用分别管理 + +> [!summary] 选哪个方案？ + +```mermaid +flowchart TD + A["需要限制并发数？"] -->|否| B["直接 go + 不管"] + A -->|是| C["errgroup.SetLimit()"] + C --> D["还需要自定义任务类型\n(结构化数据流)?"] + D -->|否| E["✅ errgroup 够用了"] + D -->|是| F["自定义协程池\nchannel + Worker"] + F --> G["还需要优雅停机?\n(关闭后拒绝新任务)?"] + G -->|是| H["golang.ccpool /\ntpool 等第三方库"] + G -->|否| E +``` + +### 五、Semaphore：另一种限速思路 + +除了 errgroup 内部用的 buffered channel + counter 方案，还有一种更直观的 semaphore 写法： + +```go +type semaphore struct { + ch chan struct{} +} + +func newSemaphore(n int) semaphore { + return semaphore{ch: make(chan struct{}, n)} +} + +func (s semaphore) acquire() { s.ch <- struct{}{} } +func (s semaphore) release() { <-s.ch } +``` + +配合 goroutine 使用的模板： + +```go +sem := newSemaphore(10) + +for _, item := range items { + go func(i Item) { + sem.acquire() + defer sem.release() + process(i) + }(item) +} +``` + +这和 errgroup 的 `SetLimit` 底层原理相同，只是手动暴露了 acquire/release。适合需要将 semaphores 传给多个独立函数的场景。 + +### 六、生产级协程池的特性要求 + +如果要在真实产品中落地自己的池子，通常需要考虑以下能力： + +| 特性 | 说明 | 实现要点 | +|------|------|---------| +| **优雅停机** | Stop 后不接收新任务，等待已提交的完成 | state machine (running/stopping/stopped)，Select 判断状态 | +| **动态扩缩容** | 根据负载调整 worker 数量 | monitor 定时检查队列长度 | +| **任务超时** | 单个任务执行超时自动跳过 | `select` + `time.After` | +| **优先级队列** | 高优先级任务先执行 | 多 channel 或 heap 结构 | +| **监控指标** | 活跃 worker、排队数、丢弃数 | atomic counters + prometheus registry | + +下面是优雅停机的核心逻辑骨架： + +```go +type State int + +const ( + StateRunning State = iota + StateStopping + StateStopped +) + +type SafePool struct { + state State + mu sync.RWMutex + queue chan Task + workers int + done chan struct{} +} + +func (p *SafePool) Submit(t Task) error { + p.mu.RLock() + defer p.mu.RUnlock() + + if p.state == StateStopped { + return ErrPoolStopped + } + + select { + case p.queue <- t: + return nil + default: + return ErrQueueFull // 队列满时非阻塞拒绝 + } +} + +func (p *SafePool) Stop() { + p.mu.Lock() + defer p.mu.Unlock() + + if p.state == StateStopped { + return + } + p.state = StateStopping + close(p.done) // 通知 worker 退出 + p.state = StateStopped +} +``` + +### 七、常见陷阱清单 + +#### 7.1 goroutine 泄漏 + +```go +// ❌ worker 永远阻塞在有缓冲的 channel 上，如果没人读 +jobs := make(chan int, 10) // 缓冲 10 +jobs <- 1 +// 如果 worker 没有启动，或者 worker 全部退出了但 channel 里还有未取的数据…… +// goroutine 就泄漏了！ + +// ✅ 始终用 WaitGroup 或 context 跟踪所有 goroutine 的生命周期 +``` + +#### 7.2 循环变量捕获（Go 1.22 之前） + +```go +// ❌ Go 1.22 之前的版本，所有 goroutine 共享同一个 i +for i := 0; i < 5; i++ { + go func() { + fmt.Println(i) // 可能全部输出 5！ + }() +} + +// ✅ 显式传参（推荐）或就地拷贝 +for i := 0; i < 5; i++ { + go func(val int) { + fmt.Println(val) + }(i) +} +``` + +> [!note] Go 1.22 变更 +> +> 从 Go 1.22 开始，循环变量的行为被修改为每次迭代创建新的副本。如果你的项目仍跑在更早的版本上，上述问题依然是真实的 bug。 + +#### 7.3 channel 未关闭导致阻塞 + +```go +// ❌ main 函数在 wg.Wait() 之前就返回了——results 永远不会被关闭 +var wg sync.WaitGroup +for w := 0; w < 3; w++ { + wg.Add(1) + go func() { + defer wg.Done() + for j := range jobs { + results <- j * 2 + } + }() +} +wg.Wait() +// 此处缺少 close(results) + +// ✅ 独立的 goroutine 负责在 wg.Wait() 后关闭 results +go func() { + wg.Wait() + close(results) +}() +``` + +#### 7.4 队列满了之后无限阻塞 + +```go +// ❌ submit 会一直阻塞直到有人消费 +p.queue <- task + +// ✅ 用 select + timeout/done channel 避免死锁 +select { +case p.queue <- task: + return nil +case <-ctx.Done(): + return ctx.Err() +default: + return ErrQueueFull +} +``` + +## 关联笔记 + +- [[Go Context]] diff --git a/hzh/REDIS/分布式限流.md b/hzh/REDIS/分布式限流.md new file mode 100644 index 0000000..10b5f89 --- /dev/null +++ b/hzh/REDIS/分布式限流.md @@ -0,0 +1,441 @@ +--- +tags: [redis, rate-limiting, distributed-system, lua-script, sliding-window, token-bucket] +create time: 2026-05-30 14:30 +--- + +# 分布式限流（基于 Redis） + +## 概述 + +假设你写了一个用户注册接口，设置了每秒最多处理 100 个请求。在单机环境下这没问题——一个计数器就够了。 + +但当你把服务扩展到 10 台机器后，问题来了：**每台机器各自计数，实际吞吐量变成了 1000 QPS**，远超数据库的承载能力。 + +> [!question]- 💡 思考：你怎么解决？ +> +> 现在有 10 个服务实例，每个都在自己的内存里维护计数器。用户的一个请求打到任意一台机器上。 +> +> 想一想——如果让你设计一个方案，让所有机器看到同一个数字，你会怎么做？ + +答案是：**把所有计数器的状态集中到一个共享存储**。Redis 天然适合这个场景——原子操作保证一致性、低延迟支撑高频判断、集群模式可以横向扩展。 + +```mermaid +flowchart LR + subgraph "多实例应用" + I1["📦 实例 A"] + I2["📦 实例 B"] + I3["📦 实例 C"] + end + + GW["🌐 API Gateway"] --> I1 + GW --> I2 + GW --> I3 + + I1 -->|共享 Key| R(("🔴 Redis
统一限流状态")) + I2 -->|共享 Key| R + I3 -->|共享 Key| R + + style R fill:#ffebee,stroke:#ef5350,stroke-width:3px,color:#000 +``` + +接下来我们从最简单的方案开始，一步步看到边界和问题，再引入更复杂的算法来解决它们。**每一节的"缺陷"都是下一节存在的理由。** + +--- + +## 一、从固定窗口开始 + +### 1.1 最简单的方法：固定窗口计数器 + +核心思想就两句话： + +1. 把时间切成一个个**等长的窗口**（比如每 1 秒一个窗口） +2. 每个窗口内用一个**计数器**记录请求数，到达阈值就拒绝 + +```mermaid +flowchart TD + Req["请求到达"] --> TS["取当前时间戳
映射到当前窗口起始点"] + TS --> Key["构造 Key:
rate:limit:{id}:{windowStart}"] + Key --> INCR["INCR 自增"] + INCR --> First{"首次?"} + First -- 是 --> Expire["EXPIRE 设过期时间"] + First -- 否 --> SkipExpire["跳过 EXPIRE"] + Expire --> Check{"count > max?"} + SkipExpire --> Check + Check -- 是 --> Reject["❌ 拒绝 (429)"] + Check -- 否 --> Allow["✅ 放行"] +``` + +对应的 Go + Lua 实现： + +```go +const fixedWindowLua = ` +local key = KEYS[1] +local maxCount = tonumber(ARGV[1]) +local window = tonumber(ARGV[2]) + +local count = redis.call('INCR', key) +if count == 1 then + redis.call('EXPIRE', key, window) +end +return count +` +``` + +调用时这样构造 Key： + +```go +// 将时间戳对齐到窗口起点，确保同一窗口共用一个 Key +windowStart := time.Now().Unix() / int64(windowSeconds) * int64(windowSeconds) +key := fmt.Sprintf("rate:limit:%s:%d", userId, windowStart) +``` + +这段代码短小精悍，看起来已经够用了。但它有一个**隐蔽的边界问题**。 + +### 1.2 边界突发：窗口的"双杀"效应 + +> [!warning]- ⚠️ 先看一个具体例子 +> +> 限制 100 请求/秒，窗口大小 1 秒。考虑两个相邻窗口的交界处： +> +> | 时刻 | 所在窗口 | 行为 | +> |------|---------|------| +> | t = 0.9s | 第 0 秒窗口 | 计数 +1，累计 100 | +> | t = 1.1s | 第 1 秒窗口 | **计数归零** +1，累计 100 | +> +> **在 0.2 秒内通过了 200 个请求**，突增量达到了限制的两倍。 + +为什么会这样？因为**前后两个窗口各自独立计数，互不关心**。窗口切换的那一刻，新旧两个窗口可以同时满负荷运行。 + +这个问题的根源在于「窗口」太粗了。我们换一个思路：**不要只盯着当前窗口，而是看最近一段时间内的全部请求。** + +这就是下一节要讲的滑动窗口。 + +--- + +## 二、更精确：滑动窗口 + +### 2.1 滑动窗口计数器 + +先别急着看代码，想想你直觉上会怎么改进固定窗口： + +> 与其让两个相邻窗口各算各的，不如用**多个小窗口加权平均**来估算当前窗口的总量？ + +比如把 1 秒分成 N=10 个子窗口，每个子窗口 100ms。当请求在第 8 个窗口时，前 7 个已经过期的窗口按剩余时间比例折算贡献度： + +```mermaid +flowchart LR + subgraph "N=3 个子窗口示意图" + W1["⬛ 旧窗口
已过期的部分 × 占比"] + W2["🟨 当前完整窗口
100% 计入"] + W3["🟩 新窗口
刚开始，从 0 累积"] + end + + Current["当前请求时刻"] --> Calc["weightedCount = W1.weight × W1.count + W2.count"] + Calc --> Compare{"total > max?"} + Compare -- 是 --> Reject["拒绝"] + Compare -- 否 --> Allow["放行"] +``` + +**效果对比：** + +| 维度 | 固定窗口 | 滑动窗口 (N=10) | +|------|---------|----------------| +| **精度** | 低（可突破 2 倍） | 中（突限量缩小到 ~1/N） | +| **空间** | O(1) per window | O(N) 子窗口 | +| **复杂度** | 极低 | 中等 | + +> [!tip] 折中选择 +> 如果你的业务对限流精度要求不高（比如日配额限流），N=10 的滑动窗口计数器已经足够，且内存开销很小。 +> +> 如果需要**逐条精确统计**，可以用 Sorted Set 实现"日志级"滑动窗口（见 2.2）。 + +### 2.2 日志级滑动窗口（Sorted Set） + +这是最精确的方案：每条请求的时间戳作为 Sorted Set 的成员（score），通过范围查询和清理来实现窗口统计。 + +**流程：** + +```mermaid +sequenceDiagram + participant App as 应用实例 + participant R as Redis + + App->>R: ZREMRANGEBYSCORE key -(now-T] now
清理过期记录 + R-->>App: 返回清理数量 + App->>R: ZCARD key
统计当前窗口内请求数 + R-->>App: currentCount + alt currentCount >= max + App->>App: ❌ 拒绝 (429) + else currentCount < max + App->>R: ZADD key now member
EXPIRE key TTL + App->>App: ✅ 放行 + end +``` + +关键是把三个步骤（清理 → 计数 → 插入）放在**一个 Lua 脚本**里，确保原子性： + +```go +const slidingWindowLua = ` +local key = KEYS[1] +local window = tonumber(ARGV[1]) -- 窗口大小 (秒) +local maxLen = tonumber(ARGV[2]) -- 最大请求数 +local now = tonumber(ARGV[3]) -- 毫秒级时间戳 +local member = ARGV[4] -- 唯一 ID (uuid / requestId) + +-- 清理窗口外的旧记录 +redis.call('ZREMRANGEBYSCORE', key, '-inf', now - window * 1000) + +-- 统计当前窗口内的数量 +local count = redis.call('ZCARD', key) + +if count >= maxLen then + return count -- 超限 +end + +-- 添加新记录 +redis.call('ZADD', key, now, member) +redis.call('EXPIRE', key, window) + +return count + 1 +` +``` + +> [!warning]- 代价与取舍 +> +> | 维度 | 滑动窗口计数器 | 滑动窗口日志 | +> |------|-------------|------------| +> | **精度** | 近似 | 精确到每一条 | +> | **空间** | O(N) 子窗口 | O(maxQPS × window) | +> | **何时不用** | — | 高并发下 sorted set 太大 | +> +> 以 maxQPS=1000、窗口 60s 为例，需要存储 **6 万条** Sorted Set 记录。对于高频 API 网关，这可能成为瓶颈——这时应该切换到令牌桶或漏桶方案。 + +--- + +## 三、换个角度：令牌桶 & 漏桶 + +前面的算法都在回答一个问题："过去一段时间内有多少请求？"但还有一种不同的思考方式——**不管输入节奏，规定系统能处理的速率。** + +### 3.1 为什么需要"桶"模型？ + +回想一下固定窗口的问题：它在窗口边界处会出现突发。令牌桶的思路完全不一样： + +> 想象一个桶，你以**恒定速度**往里灌水（补充令牌），下面的排水孔以**恒定速度**漏水（处理请求）。桶有上限，水满了就溢掉；桶空了，新来的水流不进。 + +关键在于：**桶在空闲时可以攒下多余的令牌**。这意味着短时间内的突发流量也能被容忍——只要之前积累够了就行。这对用户体验友好得多。 + +```mermaid +flowchart TD + subgraph "🪣 令牌桶内部" + TANK["令牌桶
容量 = maxBurst"] + FILL["⚡ 固定速率 r/s 持续添加令牌
最多填满到 maxBurst"] + end + + REQ["请求到来"] --> CHECK{桶中有令牌?} + CHECK -- 是 --> CONSUME["消耗 N 个令牌
✅ 放行"] + CHECK -- 否 --> REJECT["❌ 拒绝 / 等待"] + + FILL -.-> TANK + CONSUME -.->|"桶 - N"| TANK +``` + +### 3.2 令牌桶 vs 漏桶：到底有什么区别？ + +这两个概念经常混淆，核心区别只有四个字：**允许/不允许突发**。 + +| 维度 | 令牌桶 🪣 | 漏桶 🕳 | +|------|---------|-------| +| **突发** | ✅ 允许（空闲时令牌会累积） | ❌ 不允许（多余直接拒绝） | +| **输出形态** | 跟随输入节奏 | 始终匀速 | +| **隐含队列** | 无（拒绝即拒） | 有（先入桶排队再处理） | +| **保护对象** | 对客户端更友好 | 对下游更安全 | +| **典型场景** | API 网关、第三方配额 | 数据库连接池、消息队列消费者 | + +```mermaid +flowchart TD + Choice["突发流量来了怎么办？"] --> Q{"是否需要允许合理突发？"} + + Q -- "是，给客户端好体验" --> TB["🪣 选令牌桶
• REST API 限流
• 用户多次点击不都失败"] + + Q -- "否，稳定压住下游" --> LB["🕳 选漏桶
• DB 连接池限速
• 防止后端被打满"] +``` + +### 3.3 Redis + Lua 实现令牌桶 + +**为什么要用 Lua？** 因为令牌桶的操作涉及三步：读当前令牌数 → 计算补充量 → 写回新值。这三个步骤如果不原子化，多实例并发时就会超发令牌。Redis 的单线程模型 + Lua 脚本天然解决这个问题。 + +```mermaid +flowchart TD + Req["请求到达"] --> Read["读取 tokens + last_refill"] + Read --> Calc["计算补充：
newTokens = min(capacity,
tokens + elapsed × rate)"] + Calc --> Check{tokens ≥ needed?} + Check -- 是 --> Consume["tokens -= needed
✅ 放行"] + Check -- 否 --> Decline["❌ 拒绝
返还需等待时长"] + Consume --> Writeback["写回 tokens + 更新时间
设置 TTL 防僵尸 Key"] + Decline --> Writeback +``` + +完整的 Lua 脚本： + +```go +const tokenBucketLua = ` +local tokensKey = KEYS[1] +local capacity = tonumber(ARGV[1]) +local rate = tonumber(ARGV[2]) -- 每秒补充的令牌数 +local requested = tonumber(ARGV[3]) -- 请求需要的令牌数 +local now = tonumber(ARGV[4]) -- 毫秒时间戳 + +-- 读取当前状态 +local bucket = redis.call('HMGET', tokensKey, 'tokens', 'last_refill') +local tokens = tonumber(bucket[1]) +local lastFill = tonumber(bucket[2]) + +if tokens == nil then + tokens = capacity -- 首次使用：初始化为满桶 + lastFill = now +end + +-- 根据经过的时间补充令牌 +local elapsed = (now - lastFill) / 1000.0 +tokens = math.min(capacity, tokens + elapsed * rate) + +-- 尝试消费 +local allowed = 0 +local remaining = tokens +if tokens >= requested then + tokens = tokens - requested + allowed = 1 + remaining = tokens +end + +-- 写回 +redis.call('HMSET', tokensKey, 'tokens', remaining, 'last_refill', now) +redis.call('EXPIRE', tokensKey, math.ceil(capacity / rate) * 2) + +return {allowed, math.floor(remaining * 1000 / rate)} +` +``` + +对应 Go 封装： + +```go +func TokenBucketAllow(ctx context.Context, c *redis.Client, id string, capacity float64, rate float64) (bool, int64, error) { + key := fmt.Sprintf("rate:tokenbucket:%s", id) + now := time.Now().UnixMilli() + + result, err := c.Eval(ctx, tokenBucketLua, []string{key}, capacity, rate, 1, now).IntSlice() + if err != nil { + return false, 0, err + } + // result[0]: 1=放行 0=拒绝, result[1]: 下次可用毫秒数 + return result[0] == 1, result[1], nil +} +``` + +--- + +## 四、生产实践：组合与优化 + +单一种算法通常不够用。现实中的限流体系像洋葱一样层层叠加——**每层保护不同的目标，用不同的算法。** + +### 4.1 多级限流架构 + +```mermaid +flowchart TD + Req["请求进来"] --> L1 + L1["🔴 L1: IP 维度
防刷 / 防攻击"] -- pass --> L2 + L2["🟡 L2: 用户维度
控制每日/每月配额"] -- pass --> L3 + L3["🟢 L3: 接口维度
保护服务峰值"] -- pass --> Biz["执行业务逻辑"] + + L1 -- fail --> R["返回 429"] + L2 -- fail --> R + L3 -- fail --> R + + style R fill:#ffebee,color:#000 +``` + +每层的典型配置： + +| 层级 | 限流目标 | 推荐算法 | 典型参数 | +|------|---------|---------|---------| +| **L1 IP 层** | 防暴力扫描、恶意攻击 | 短窗口固定计数 | 100 req/s/IP | +| **L2 用户层** | 控制付费/免费配额 | 天级滑动窗口日志 | 1000 req/day | +| **L3 接口层** | 保护实例不被打满 | 令牌桶 | 5000 req/s/endpoint | + +### 4.2 优雅降级：被限流时该说什么 + +被限流的请求不应静默丢弃，而应给出明确信号： + +```http +HTTP/1.1 429 Too Many Requests +Retry-After: 2 # 建议客户端等待的秒数 +X-RateLimit-Limit: 100 # 限流阈值 +X-RateLimit-Remaining: 0 # 当前窗口剩余配额 +X-RateLimit-Reset: 1690000060 # 窗口重置时间戳 + +{ + "error": { + "code": "RATE_LIMITED", + "message": "请求过于频繁，请稍后再试", + "retry_after_seconds": 2 + } +} +``` + +> [!tip]- 客户端收到 429 后该怎么做？ +> +> 1. **遵守 `Retry-After` 头部**进行等待重试 +> 2. 使用**指数退避 + jitter**：`wait = min(base × 2^attempt + random_jitter, Retry-After)` +> 3. **不要**立即无脑重发——这会造成 "惊群效应"（thundering herd），再次打垮恢复中的服务 + +### 4.3 性能优化清单 + +```mermaid +flowchart LR + Pipelines["Pipeline 减少 RTT
MULTI/EXEC 合并命令"] + EvalSHA["EVALSHA 预加载 Lua
避免重复传输脚本"] + HashKey["Hash 压缩 Key 数
HSET 替代多个独立 Key"] + SCAN["SCAN 替代 KEYS
避免阻塞其他客户端"] + + Pipelines --> Perf["⚡ 整体性能"] + EvalSHA --> Perf + HashKey --> Perf + SCAN --> Perf +``` + +| 技巧 | 收益 | 注意 | +|------|------|------| +| **Pipeline** | 降低网络往返，QPS 翻倍 | 注意批量命令不能拆成单独事务 | +| **EVALSHA** | 节省带宽，Lua 调用提速 | 需先用 SCRIPT LOAD 预加载脚本 | +| **Hash 聚合 Key** | 减少内存碎片和过期管理开销 | 设计时就要规划好聚合粒度 | +| **SCAN 替代 KEYS** | 避免大库扫按时阻塞服务端 | KEYS 在数据量大时影响严重 | + +--- + +## 五、排错指南 + +> [!example]- 出了问题先从这些方向排查 + +| 症状 | 可能原因 | 排查方法 | +|------|---------|---------| +| 超出限制大量通过 | 时钟不同步导致 Lua 中 `now` 异常 | `date` 比对 Redis 和应用服务器 | +| Redis CPU 飙高 | 滑动窗口日志 Sorted Set 规模过大 | `MEMORY USAGE key` 查大 Key；换令牌桶 | +| 偶发性限流不一致 | Pipeline 非原子性拆分了判断逻辑 | 确认核心判断走单个 Lua 脚本 | +| 内存持续增长 | Key 的 EXPIRE 未生效 | `TTL key` 检查是否已过期 | +| Cluster 下限流失效 | Key 跨 slot，Lua 无法多 Key 操作 | 使用 Hash Tag `{user123}.daily` 锁定 slot | + +> [!warning]- Redis Cluster 的关键陷阱 +> Lua 脚本中涉及多个 Key 时，它们必须在**同一个 hash slot**。默认的 Key 格式可能导致不同实例落在不同 slot 上。 +> +> **解决方法：** 用 `{}` 包裹路由键，强制 Hash Tag 一致： +> ``` +> rate:limit:{user123}:daily ← {} 保证路由到同一 slot +> rate:limit:{user123}:hourly +> ``` + +--- + +## 关联笔记 + +- [[02-服务治理/06-容错模式]] — 限流是容错体系的第三层防线 +- [[02-服务治理/01-API网关]] — API 网关层的限流插件集成