第14章 · 服务治理：熔断、限流、降级（正式版）

// pkg/circuitbreaker/simple.go
package circuitbreaker

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

// State 熔断器状态
type State int

const (
	StateClosed State = iota // 闭合：正常放行
	StateOpen                // 打开：快速失败
	StateHalfOpen            // 半开：放少量试探
)

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

// Breaker 简单熔断器
type Breaker struct {
	mu               sync.Mutex
	state            State
	failureThreshold int           // 错误数阈值：攒够多少个失败就跳闸
	failureCount     int           // 当前已累计的失败数
	lastStateChange  time.Time     // 上次状态变更时间
	openTimeout      time.Duration // Open 状态持续多久后转 HalfOpen
	halfOpenMaxCalls int           // HalfOpen 最多放几个试探请求
	halfOpenCalls    int           // HalfOpen 已放行的试探数
}

func New(failureThreshold int, openTimeout time.Duration) *Breaker {
	return &Breaker{
		state:            StateClosed,
		failureThreshold: failureThreshold,
		openTimeout:      openTimeout,
		halfOpenMaxCalls: 5, // 默认放 5 个试探请求
	}
}

// State 返回当前状态（供外部观察状态机，如降级判断、打印调试）
func (b *Breaker) State() State {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.state
}

// pkg/circuitbreaker/simple.go（续）
// Allow 判断请求是否允许通过
func (b *Breaker) Allow() error {
	b.mu.Lock()
	defer b.mu.Unlock()

	switch b.state {
	case StateClosed:
		// 步骤 1：闭合态直接放行
		return nil
	case StateOpen:
		// 步骤 2：开放态，只有超时后才转半开并放行一个试探
		if time.Since(b.lastStateChange) > b.openTimeout {
			b.state = StateHalfOpen
			b.halfOpenCalls = 0
			return nil
		}
		return ErrCircuitOpen
	case StateHalfOpen:
		// 步骤 3：半开态限制试探请求数，超了就拒绝
		if b.halfOpenCalls < b.halfOpenMaxCalls {
			b.halfOpenCalls++
			return nil
		}
		return ErrCircuitOpen
	}
	return nil
}

// pkg/circuitbreaker/simple.go（续）
// MarkSuccess 标记一次成功
func (b *Breaker) MarkSuccess() {
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.state == StateHalfOpen {
		// 半开态成功 → 转 Closed，统计清零（失败计数和试探名额都要归零）
		b.state = StateClosed
		b.failureCount = 0
		b.halfOpenCalls = 0
	}
}

// MarkFailure 标记一次失败
func (b *Breaker) MarkFailure() {
	b.mu.Lock()
	defer b.mu.Unlock()

	if b.state == StateHalfOpen {
		// 半开态失败 → 立刻转 Open，重新计时
		b.state = StateOpen
		b.lastStateChange = time.Now()
		return
	}

	// 闭合态：累计失败数，攒够阈值就跳闸
	b.failureCount++
	if b.failureCount >= b.failureThreshold {
		b.state = StateOpen
		b.lastStateChange = time.Now()
	}
}

// cmd/breaker_demo/main.go
package main

import (
	"fmt"
	"time"

	"content-community/pkg/circuitbreaker"
)

func main() {
	// 阈值 3 个失败，跳闸后 1 秒进入半开
	b := circuitbreaker.New(3, time.Second)

	// 阶段 1：制造 3 次失败，触发跳闸
	for i := 0; i < 3; i++ {
		_ = b.Allow()
		b.MarkFailure()
	}
	fmt.Println("3 次失败后:", b.State()) // StateOpen(1)

	// 阶段 2：熔断打开期间，请求直接被拒
	err := b.Allow()
	fmt.Println("熔断期间调用:", err) // circuit breaker is open

	// 阶段 3：等 1 秒超时窗口过去，进入半开并试探成功
	time.Sleep(1100 * time.Millisecond)
	_ = b.Allow()          // 这次放行（转 HalfOpen + 占用试探名额）
	b.MarkSuccess()        // 试探成功 → 回 Closed
	fmt.Println("试探成功后:", b.State()) // StateClosed(0)

	// 阶段 4：恢复后请求正常放行
	fmt.Println("恢复后调用:", b.Allow()) // <nil>
}

go run ./cmd/breaker_demo/

3 次失败后: 1
熔断期间调用: circuit breaker is open
试探成功后: 0
恢复后调用: <nil>

// pkg/ratelimit/counter.go
package counter

import "sync"

// CounterLimiter 计数器限流器，限制并发数
type CounterLimiter struct {
	mu       sync.Mutex
	current  int // 当前并发数
	maxConns int // 最大并发数
}

func New(maxConns int) *CounterLimiter {
	return &CounterLimiter{maxConns: maxConns}
}

// Allow 尝试获取一个并发槽位
func (l *CounterLimiter) Allow() bool {
	l.mu.Lock()
	defer l.mu.Unlock()
	// 步骤 1：已达上限直接拒绝
	if l.current >= l.maxConns {
		return false
	}
	// 步骤 2：占用一个槽位
	l.current++
	return true
}

// Release 处理完毕后释放槽位（必须 defer 调用！）
func (l *CounterLimiter) Release() {
	l.mu.Lock()
	defer l.mu.Unlock()
	if l.current > 0 {
		l.current--
	}
}

// internal/service/comment.go（节选）
func (s *CommentService) Publish(ctx context.Context, req *PublishReq) error {
	if !s.limiter.Allow() {
		return errors.New("系统繁忙，请稍后再试") // 满了直接拒绝
	}
	defer s.limiter.Release() // 关键：无论如何都要释放
	return s.repo.Insert(ctx, req)
}

// pkg/ratelimit/fixedwindow.go
package fixedwindow

import (
	"sync"
	"time"
)

// FixedWindowLimiter 固定窗口限流器
type FixedWindowLimiter struct {
	mu        sync.Mutex
	window    time.Duration // 窗口大小，如 1s
	threshold int           // 每个窗口内允许的最大请求数
	count     int           // 当前窗口已放行的请求数
	windowEnd time.Time     // 当前窗口的结束时间
}

func New(window time.Duration, threshold int) *FixedWindowLimiter {
	return &FixedWindowLimiter{
		window:    window,
		threshold: threshold,
		windowEnd: time.Now().Add(window),
	}
}

// Allow 判断是否放行
func (l *FixedWindowLimiter) Allow() bool {
	l.mu.Lock()
	defer l.mu.Unlock()
	now := time.Now()
	// 步骤 1：跨过窗口边界，就重开一个窗口，计数清零
	if now.After(l.windowEnd) {
		l.count = 0
		l.windowEnd = now.Add(l.window)
	}
	// 步骤 2：本窗口计数已达阈值，拒绝
	if l.count >= l.threshold {
		return false
	}
	// 步骤 3：放行并计数
	l.count++
	return true
}

// pkg/ratelimit/slidingwindow.go
package slidingwindow

import (
	"sync"
	"time"
)

// SlidingWindowLimiter 滑动窗口限流器（朴素版：存所有请求的时间戳）
type SlidingWindowLimiter struct {
	mu        sync.Mutex
	window    time.Duration // 窗口大小，如 1s
	threshold int           // 窗口内允许的最大请求数
	requests  []time.Time   // 窗口内的请求时间戳
}

func New(window time.Duration, threshold int) *SlidingWindowLimiter {
	return &SlidingWindowLimiter{
		window:    window,
		threshold: threshold,
	}
}

// Allow 检查是否允许通过
func (l *SlidingWindowLimiter) Allow() bool {
	l.mu.Lock()
	defer l.mu.Unlock()

	now := time.Now()
	// 步骤 1：清理窗口外的旧请求（时间戳早于「now - window」的都过期了）
	boundary := now.Add(-l.window)
	idx := 0
	for ; idx < len(l.requests); idx++ {
		if l.requests[idx].After(boundary) {
			break
		}
	}
	l.requests = l.requests[idx:]

	// 步骤 2：最近 window 内请求数已达阈值，拒绝
	if len(l.requests) >= l.threshold {
		return false
	}
	// 步骤 3：记录本次请求并放行
	l.requests = append(l.requests, now)
	return true
}

// pkg/ratelimit/tokenbucket.go（入门版）
package tokenbucket

import (
	"sync"
	"time"
)

// TokenBucket 令牌桶限流器
type TokenBucket struct {
	mu         sync.Mutex
	capacity   int64     // 桶容量（最多攒多少令牌）
	tokens     int64     // 当前令牌数
	rate       int64     // 每秒生成的令牌数
	lastRefill time.Time // 上次补充令牌的时间
}

func New(capacity, rate int64) *TokenBucket {
	return &TokenBucket{
		capacity:   capacity,
		tokens:     capacity, // 启动时桶是满的，允许初始突发
		rate:       rate,
		lastRefill: time.Now(),
	}
}

// Allow 尝试获取一个令牌
func (tb *TokenBucket) Allow() bool {
	tb.mu.Lock()
	defer tb.mu.Unlock()

	// 步骤 1：按经过时间补充令牌（懒计算：不用后台协程，来请求时才算账）
	now := time.Now()
	elapsed := now.Sub(tb.lastRefill).Seconds()
	refill := int64(elapsed * float64(tb.rate))
	if refill > 0 {
		tb.tokens += refill
		if tb.tokens > tb.capacity {
			tb.tokens = tb.capacity // 不能超过桶容量
		}
		tb.lastRefill = now
	}

	// 步骤 2：有令牌就消费并放行，否则拒绝
	if tb.tokens > 0 {
		tb.tokens--
		return true
	}
	return false
}

// internal/middleware/ratelimit.go（用官方库的极简示例）
import "golang.org/x/time/rate"

limiter := rate.NewLimiter(rate.Limit(100), 200) // 速率 100/s，桶容量 200

if !limiter.Allow() {
	// 拒绝请求
}

// pkg/ratelimit/leakybucket.go
package leakybucket

import (
	"sync"
	"time"
)

// LeakyBucket 漏桶限流器：不管请求什么时候来，放行速率绝对均匀
type LeakyBucket struct {
	mu       sync.Mutex
	interval time.Duration // 相邻两次放行的最小间隔 = 1 / rate
	lastLeak time.Time     // 上一次放行（"滴水"）的时间
}

func New(ratePerSec int64) *LeakyBucket {
	return &LeakyBucket{
		interval: time.Second / time.Duration(ratePerSec),
	}
}

// Allow 尝试放行一个请求
func (lb *LeakyBucket) Allow() bool {
	lb.mu.Lock()
	defer lb.mu.Unlock()
	now := time.Now()
	// 步骤 1：距离上次放行还没满一个间隔，说明"滴"得太快，拒绝
	if now.Sub(lb.lastLeak) < lb.interval {
		return false
	}
	// 步骤 2：放行，并记录本次放行时间
	lb.lastLeak = now
	return true
}

// pkg/middleware/logging.go
package middleware

import (
	"context"
	"log"
	"time"

	"google.golang.org/grpc"
)

// LogInterceptor 简单日志拦截器
func LogInterceptor(
	ctx context.Context,
	req any,
	info *grpc.UnaryServerInfo,
	handler grpc.UnaryHandler,
) (any, error) {
	// 前置逻辑：记录调用开始
	start := time.Now()
	// 调用真正的 handler（业务代码在这里执行）
	resp, err := handler(ctx, req)
	// 后置逻辑：记录耗时和错误
	log.Printf("method=%s cost=%v err=%v", info.FullMethod, time.Since(start), err)
	return resp, err
}

// cmd/server/main.go（节选）
server := grpc.NewServer(
	grpc.ChainUnaryInterceptor(
		middleware.LogInterceptor,
		// 限流、熔断拦截器往下排……
	),
)

// pkg/middleware/ratelimit.go
package middleware

import (
	"context"

	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

// Limiter 限流器接口（跟写部分会实现它）
type Limiter interface {
	Allow(key string) bool
	Release(key string)
}

// GRPCLimitInterceptor 全应用统一限流
func GRPCLimitInterceptor(l Limiter) grpc.UnaryServerInterceptor {
	return func(ctx context.Context, req any,
		info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
		// 步骤 1：用固定 key 对整个应用一起限流
		const key = "global"
		if !l.Allow(key) {
			return nil, status.Error(codes.ResourceExhausted, "rate limited")
		}
		// 步骤 2：处理完释放槽位
		defer l.Release(key)
		return handler(ctx, req)
	}
}

// pkg/middleware/ratelimit.go（续）
// ServiceLimitInterceptor 按 FullMethod 限流，粒度更细
func ServiceLimitInterceptor(l Limiter) grpc.UnaryServerInterceptor {
	return func(ctx context.Context, req any,
		info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
		// 步骤 1：用方法名作为 key，针对每个方法独立限流
		if !l.Allow(info.FullMethod) {
			return nil, status.Error(codes.ResourceExhausted, "rate limited")
		}
		// 步骤 2：处理完释放
		defer l.Release(info.FullMethod)
		return handler(ctx, req)
	}
}

// pkg/middleware/ratelimit.go（续）
// uidRequest 约定：实现了 GetUid 的请求都能被业务级限流
type uidRequest interface {
	GetUid() int64
}

// BizLimitInterceptor 按业务字段（如 UID）限流
func BizLimitInterceptor(l Limiter) grpc.UnaryServerInterceptor {
	return func(ctx context.Context, req any,
		info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
		// 步骤 1：类型断言，取出 UID 作为 key
		if r, ok := req.(uidRequest); ok {
			key := strconv.FormatInt(r.GetUid(), 10)
			if !l.Allow(key) {
				return nil, status.Error(codes.ResourceExhausted, "rate limited")
			}
			// 步骤 2：处理完释放
			defer l.Release(key)
		}
		// 没实现 GetUid 的请求直接放行
		return handler(ctx, req)
	}
}

// internal/service/user_service_ratelimited.go
// RateLimitedUserService 用装饰器模式给 UserService 加限流
type RateLimitedUserService struct {
	UserServiceServer            // 内嵌原始实现（继承它的所有方法）
	limiter           Limiter
}

func (r *RateLimitedUserService) GetProfile(ctx context.Context,
	req *GetProfileReq) (*ProfileResp, error) {
	// 步骤 1：按 UID 限流
	key := strconv.FormatInt(req.GetUid(), 10)
	if !r.limiter.Allow(key) {
		return nil, status.Error(codes.ResourceExhausted, "rate limited")
	}
	// 步骤 2：放行到被装饰的原始实现
	defer r.limiter.Release(key)
	return r.UserServiceServer.GetProfile(ctx, req)
}

// pkg/middleware/circuitbreaker.go
package middleware

import (
	"context"

	"github.com/go-kratos/aegis/circuitbreaker"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

// CircuitBreakerInterceptor 熔断拦截器
func CircuitBreakerInterceptor(cb *circuitbreaker.Breaker) grpc.UnaryServerInterceptor {
	return func(ctx context.Context, req any,
		info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
		// 步骤 1：Allow 返回 error 表示熔断已开启，直接拒绝
		if err := cb.Allow(); err != nil {
			return nil, status.Error(codes.Unavailable, "circuit breaker open")
		}
		// 步骤 2：调用业务，拿到结果与错误
		resp, err := handler(ctx, req)
		// 步骤 3：关键——成对上报成功/失败，否则熔断状态永不更新
		if err != nil {
			cb.MarkFailed()
		} else {
			cb.MarkSuccess()
		}
		return resp, err
	}
}

// internal/service/user_service.go（节选）
// 业务代码里结合限流器状态做降级
func (s *UserService) GetProfile(ctx context.Context, uid int64) (*Profile, error) {
	// 步骤 1：已触发限流时，只查缓存，不碰 DB
	if s.limiter.IsTriggered() {
		p, err := s.cache.Get(ctx, uid)
		if err == nil {
			return p, nil
		}
		// 缓存也未命中，降级到底：返回错误
		return nil, fmt.Errorf("get profile: 限流降级且缓存未命中") // 生产中应返回项目统一的限流错误码，供上层识别
	}
	// 步骤 2：正常路径：查缓存，未命中查 DB 并回写
	return s.cacheAside.Get(ctx, uid)
}

// internal/client/user.go（节选）
ctx, cancel := context.WithTimeout(ctx, time.Second)
defer cancel() // 不 cancel 会泄漏 context 资源
resp, err := client.GetProfile(ctx, req)

// pkg/middleware/logging.go（完整版思路）
func LogInterceptor(ctx context.Context, req any,
	info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
	// 步骤 1：记录开始时间
	start := time.Now()
	// 步骤 2：调用 handler
	resp, err := handler(ctx, req)
	// 步骤 3：记录耗时、错误码、方法名
	log.Printf("method=%s cost=%v err=%v", info.FullMethod, time.Since(start), err)
	return resp, err
}

// pkg/circuitbreaker/circuitbreaker.go
package circuitbreaker

import (
	"errors"
	"sync"
	"time"

	"go.uber.org/zap"
)

// State 熔断器三态（原理见 3.2 节）
type State int32

const (
	StateClosed   State = iota // 关闭：正常放行，统计失败率
	StateOpen                  // 打开：快速失败，不发请求
	StateHalfOpen              // 半开：放少量试探请求
)

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

// Config 配置项（注释为推荐默认值）
type Config struct {
	Window, OpenTimeout      time.Duration // 统计窗口 / Open 到期转 HalfOpen，默认 10s / 30s
	MinRequests, HalfOpenMax int           // 最小请求数 / 半开试探数，默认 20 / 5
	FailureRate              float64       // 失败率阈值：0.5 表示 50%
}

const bucketCount = 10 // 滑动窗口切成 10 个小时间片，环形复用

type bucket struct{ success, failure int64 }

// pkg/circuitbreaker/circuitbreaker.go（续）
// CircuitBreaker 一个 service+method 维度一个实例
type CircuitBreaker struct {
	mu           sync.Mutex
	cfg          Config
	state        State
	buckets      [bucketCount]bucket // 环形分桶的滑动窗口
	bucketSize   time.Duration       // 小时间片时长 = Window/10
	curBucket    int                 // 当前时间片下标
	lastTime     time.Time           // 窗口已推进到的时间
	openedAt     time.Time           // 进入 Open 的时刻
	halfOpenUsed int                 // HalfOpen 已放行的试探数
	logger       *zap.Logger
}

// New 创建熔断器；logger 传 nil 则不打日志
func New(name string, cfg Config, logger *zap.Logger) *CircuitBreaker {
	if logger == nil {
		logger = zap.NewNop()
	}
	return &CircuitBreaker{cfg: cfg, bucketSize: cfg.Window / bucketCount,
		lastTime: time.Now(), logger: logger.With(zap.String("breaker", name))}
}

// State 返回当前状态，给降级判断、指标上报用
func (cb *CircuitBreaker) State() State {
	cb.mu.Lock()
	defer cb.mu.Unlock()
	return cb.state
}

// pkg/circuitbreaker/circuitbreaker.go（续）
// advance 把窗口推进到 now，顺路清零"路过"的旧时间片
func (cb *CircuitBreaker) advance(now time.Time) {
	steps := int(now.Sub(cb.lastTime) / cb.bucketSize)
	if steps <= 0 {
		return
	}
	if steps >= bucketCount { // 距上次请求超过整个窗口：全部过期
		cb.buckets = [bucketCount]bucket{}
	} else {
		for i := 1; i <= steps; i++ {
			cb.buckets[(cb.curBucket+i)%bucketCount] = bucket{}
		}
	}
	cb.curBucket = (cb.curBucket + steps) % bucketCount
	cb.lastTime = cb.lastTime.Add(time.Duration(steps) * cb.bucketSize)
}

// pkg/circuitbreaker/circuitbreaker.go（续）
// Allow 调用前检查：返回 nil 放行；返回 ErrCircuitOpen 直接快速失败
func (cb *CircuitBreaker) Allow() error {
	cb.mu.Lock()
	defer cb.mu.Unlock()
	switch cb.state {
	case StateClosed:
		return nil
	case StateOpen:
		if time.Since(cb.openedAt) < cb.cfg.OpenTimeout {
			return ErrCircuitOpen // 未到期：继续快速失败
		}
		cb.state = StateHalfOpen // 到期：转半开，开始试探
		cb.halfOpenUsed = 0
		cb.logger.Info("circuit: Open -> HalfOpen")
		fallthrough
	case StateHalfOpen:
		if cb.halfOpenUsed >= cb.cfg.HalfOpenMax { // 试探名额有限
			return ErrCircuitOpen
		}
		cb.halfOpenUsed++
		return nil
	}
	return nil
}

// pkg/circuitbreaker/circuitbreaker.go（续）
// MarkSuccess / MarkFailure 拿到调用结果后成对上报
func (cb *CircuitBreaker) MarkSuccess() { cb.mark(true) }
func (cb *CircuitBreaker) MarkFailure() { cb.mark(false) }

func (cb *CircuitBreaker) mark(success bool) {
	cb.mu.Lock()
	defer cb.mu.Unlock()
	switch cb.state {
	case StateHalfOpen: // 一次成功即恢复，一次失败立刻回 Open
		if success {
			cb.toState(StateClosed, "probe success")
		} else {
			cb.toState(StateOpen, "probe failure")
		}
		return
	case StateOpen:
		return // 被拒的请求没真正发出，不参与统计（3.2 节的坑 2）
	}
	cb.advance(time.Now())
	if success {
		cb.buckets[cb.curBucket].success++
	} else {
		cb.buckets[cb.curBucket].failure++
	}
	cb.checkTrip()
}

// checkTrip Closed 态判定：样本足够且失败率达标才熔断
func (cb *CircuitBreaker) checkTrip() {
	var total, failure int64
	for _, b := range cb.buckets {
		total += b.success + b.failure
		failure += b.failure
	}
	if total < cb.cfg.MinRequests {
		return // 样本不足不判定，防止小流量误熔断
	}
	if float64(failure)/float64(total) >= cb.cfg.FailureRate {
		cb.toState(StateOpen, "failure rate over threshold")
	}
}

// toState 状态迁移统一入口：清空统计并打日志
func (cb *CircuitBreaker) toState(s State, reason string) {
	cb.state = s
	cb.halfOpenUsed = 0
	cb.buckets = [bucketCount]bucket{}
	if s == StateOpen {
		cb.openedAt = time.Now()
		cb.logger.Warn("circuit: -> Open", zap.String("reason", reason))
	} else {
		cb.logger.Info("circuit: -> Closed", zap.String("reason", reason))
	}
}

// pkg/circuitbreaker/circuitbreaker.go（续）
// Group 按 service+method 维度管理一组熔断器
type Group struct {
	mu       sync.Mutex
	cfg      Config
	logger   *zap.Logger
	breakers map[string]*CircuitBreaker
}

func NewGroup(cfg Config, logger *zap.Logger) *Group {
	return &Group{cfg: cfg, logger: logger, breakers: map[string]*CircuitBreaker{}}
}

// Get 取指定维度的熔断器，不存在则创建（配置组内共享）
func (g *Group) Get(name string) *CircuitBreaker {
	g.mu.Lock()
	defer g.mu.Unlock()
	if cb, ok := g.breakers[name]; ok {
		return cb
	}
	g.breakers[name] = New(name, g.cfg, g.logger)
	return g.breakers[name]
}

go build ./pkg/circuitbreaker/

// pkg/ratelimit/tokenbucket.go
package ratelimit

import (
	"sync"
	"time"

	"go.uber.org/zap"
)

// tokenBucket 单个令牌桶（懒计算版：无需后台协程补令牌）
type tokenBucket struct {
	mu         sync.Mutex
	tokens     float64   // 当前令牌数；浮点数才能支持速率 < 1 个/秒
	lastRefill time.Time // 上次补充令牌的时间
}

// TokenBucketLimiter 按 key 维护独立令牌桶，key 就是限流维度
type TokenBucketLimiter struct {
	buckets        sync.Map // key -> *tokenBucket
	capacity, rate float64  // 桶容量（决定可突发量）/ 每秒补充数（决定长期 QPS）
	logger         *zap.Logger
}

func NewTokenBucketLimiter(capacity, ratePerSec int, logger *zap.Logger) *TokenBucketLimiter {
	if logger == nil {
		logger = zap.NewNop()
	}
	return &TokenBucketLimiter{capacity: float64(capacity),
		rate: float64(ratePerSec), logger: logger}
}

// Allow 尝试为 key 取 1 个令牌：取到放行，取不到拒绝
func (l *TokenBucketLimiter) Allow(key string) bool {
	b, _ := l.buckets.LoadOrStore(key,
		&tokenBucket{tokens: l.capacity, lastRefill: time.Now()}) // 新桶初始满，允许冷启动突发
	tb := b.(*tokenBucket)
	tb.mu.Lock()
	defer tb.mu.Unlock()
	now := time.Now()
	tb.tokens += now.Sub(tb.lastRefill).Seconds() * l.rate // 懒计算：应补令牌 = 时间差 × 速率
	if tb.tokens > l.capacity {
		tb.tokens = l.capacity // 桶满，多余令牌丢弃
	}
	tb.lastRefill = now
	if tb.tokens >= 1 {
		tb.tokens--
		return true
	}
	l.logger.Warn("ratelimit: rejected", zap.String("key", key))
	return false
}

// pkg/ratelimit/slidingwindow.go
package ratelimit

import (
	"sync"
	"time"

	"go.uber.org/zap"
)

// slidingWindow 大窗口切成 10 个小时间片，各自计数
type slidingWindow struct {
	mu       sync.Mutex
	counts   [10]int64
	cur      int       // 当前时间片下标
	lastTime time.Time // 窗口已推进到的时间
}

// SlidingWindowLimiter 按 key 维护独立滑动窗口
type SlidingWindowLimiter struct {
	windows   sync.Map
	window    time.Duration // 总窗口大小，如 1s
	threshold int64         // 窗口内允许的最大请求数
	logger    *zap.Logger
}

func NewSlidingWindowLimiter(window time.Duration, threshold int64,
	logger *zap.Logger) *SlidingWindowLimiter {
	if logger == nil {
		logger = zap.NewNop()
	}
	return &SlidingWindowLimiter{window: window, threshold: threshold, logger: logger}
}

// Allow 判断 key 本次请求是否放行（窗口推进与熔断器 advance 同一思路）
func (l *SlidingWindowLimiter) Allow(key string) bool {
	w, _ := l.windows.LoadOrStore(key, &slidingWindow{lastTime: time.Now()})
	sw := w.(*slidingWindow)
	sw.mu.Lock()
	defer sw.mu.Unlock()
	now, slot := time.Now(), l.window/10 // slot = 小时间片时长
	if steps := int(now.Sub(sw.lastTime) / slot); steps > 0 {
		if steps >= 10 {
			sw.counts = [10]int64{}
		} else {
			for i := 1; i <= steps; i++ {
				sw.counts[(sw.cur+i)%10] = 0
			}
		}
		sw.cur = (sw.cur + steps) % 10
		sw.lastTime = sw.lastTime.Add(time.Duration(steps) * slot)
	}
	var total int64 // 整个窗口内的总请求数
	for _, c := range sw.counts {
		total += c
	}
	if total >= l.threshold {
		l.logger.Warn("ratelimit: rejected", zap.String("key", key))
		return false
	}
	sw.counts[sw.cur]++
	return true
}

// pkg/ratelimit/ratelimit_test.go
package ratelimit

import "testing"

func TestTokenBucket(t *testing.T) {
	// 容量 5、速率 100/s：初始能连过 5 个，第 6 个被拒
	l := NewTokenBucketLimiter(5, 100, nil)
	for i := 0; i < 5; i++ {
		if !l.Allow("global") {
			t.Fatalf("第 %d 个请求不应被拒绝", i+1)
		}
	}
	if l.Allow("global") {
		t.Fatal("第 6 个请求应被拒绝")
	}
}

func TestSlidingWindow(t *testing.T) {
	// 窗口 1s、阈值 3：连发 3 个通过，第 4 个被拒
	l := NewSlidingWindowLimiter(1e9, 3, nil)
	for i := 0; i < 3; i++ {
		if !l.Allow("global") {
			t.Fatalf("第 %d 个请求不应被拒绝", i+1)
		}
	}
	if l.Allow("global") {
		t.Fatal("第 4 个请求应被拒绝")
	}
}

go test ./pkg/ratelimit/ -v

=== RUN   TestTokenBucket
--- PASS: TestTokenBucket (0.00s)
=== RUN   TestSlidingWindow
--- PASS: TestSlidingWindow (0.00s)
PASS
ok  	content-community/pkg/ratelimit

// pkg/circuitbreaker/interceptor.go
package circuitbreaker

import (
	"context"

	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

// GRPCClientInterceptor gRPC 客户端熔断拦截器（按 FullMethod 维度）
func GRPCClientInterceptor(g *Group) grpc.UnaryClientInterceptor {
	return func(ctx context.Context, method string, req, reply any,
		cc *grpc.ClientConn, invoker grpc.UnaryInvoker,
		opts ...grpc.CallOption) error {
		cb := g.Get(method) // 一个 service+method 一个熔断器
		// 步骤 1：调用前检查，Open 就地快速失败，不发网络请求
		if err := cb.Allow(); err != nil {
			return status.Error(codes.Unavailable, "circuit breaker open: "+method)
		}
		// 步骤 2：真正发起 RPC；步骤 3：按错误类别上报统计
		err := invoker(ctx, method, req, reply, cc, opts...)
		cb.report(err)
		return err
	}
}

// report 错误分类上报：只有"下游故障类"错误才触发熔断统计
func (cb *CircuitBreaker) report(err error) {
	if err == nil {
		cb.MarkSuccess()
		return
	}
	switch status.Code(err) {
	case codes.DeadlineExceeded, codes.Unavailable, codes.Canceled, // 超时/连不上/取消
		codes.Internal, codes.Unknown: // 服务端内部错误等 5xx 类
		cb.MarkFailure()
	default: // InvalidArgument / NotFound 等业务错：调用本身是成功的
		cb.MarkSuccess()
	}
}

go build ./pkg/circuitbreaker/ && go test ./pkg/circuitbreaker/ -v

// pkg/degradation/degradation.go
package degradation

import (
	"context"
	"sync"

	"go.uber.org/zap"
)

// Strategy 降级策略接口：每个可降级的方法实现一份
type Strategy interface {
	ShouldDegrade() bool                                // 当前是否需要降级（如：熔断器已打开）
	Fallback(ctx context.Context, req any) (any, error) // 兜底：返回默认值/缓存/简化数据
}

// Manager 降级管理器：按方法注册策略、执行带降级的调用
type Manager struct {
	mu         sync.RWMutex // 注册只发生在启动期，读多写少
	strategies map[string]Strategy
	logger     *zap.Logger
}

func NewManager(logger *zap.Logger) *Manager {
	return &Manager{strategies: map[string]Strategy{}, logger: logger}
}

// Register 给方法注册降级策略（也可接配置中心做一键降级开关）
func (m *Manager) Register(method string, s Strategy) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.strategies[method] = s
}

// Do 执行一次带降级的调用：事前判定要降级，或正常调用失败，都走 Fallback
func (m *Manager) Do(ctx context.Context, method string, req any,
	call func(context.Context, any) (any, error)) (any, error) {
	m.mu.RLock()
	s := m.strategies[method]
	m.mu.RUnlock()
	if s != nil && s.ShouldDegrade() {
		m.logger.Warn("degradation: triggered", zap.String("method", method))
		return s.Fallback(ctx, req)
	}
	resp, err := call(ctx, req)
	if err == nil || s == nil {
		return resp, err // 调用成功，或没注册策略：原样返回
	}
	m.logger.Warn("degradation: fallback on error",
		zap.String("method", method), zap.Error(err))
	return s.Fallback(ctx, req)
}

// internal/service/post_service.go（节选）
type Author struct{ Nickname, Avatar string }

// AuthorFallback 作者服务的降级策略：熔断打开就不查了，返回占位作者
type AuthorFallback struct {
	breaker *circuitbreaker.CircuitBreaker // 复用第 1 步的熔断器状态
}

func (f *AuthorFallback) ShouldDegrade() bool {
	return f.breaker.State() == circuitbreaker.StateOpen
}

func (f *AuthorFallback) Fallback(_ context.Context, _ any) (any, error) {
	return &Author{Nickname: "未知用户", Avatar: "/img/avatar-default.png"}, nil
}

// GetPostList 作者信息用 degrade.Do 包一层；启动时注册：
// degrade.Register("user.GetAuthor",
//   &AuthorFallback{breaker: cbGroup.Get("/user.v1.UserService/GetAuthor")})
func (s *PostService) GetPostList(ctx context.Context) ([]*Post, error) {
	posts, err := s.repo.List(ctx)
	if err != nil {
		return nil, fmt.Errorf("list posts: %w", err)
	}
	for _, p := range posts {
		v, err := s.degrade.Do(ctx, "user.GetAuthor", p.AuthorID,
			func(ctx context.Context, req any) (any, error) {
				return s.userClient.GetAuthor(ctx, req.(int64)) // 正常路径
			})
		if err == nil {
			p.Author = v.(*Author)
		}
	}
	return posts, nil
}

content-community/
├── configs/config.yaml             # 追加 governance.circuitbreaker 配置
└── pkg/
    ├── circuitbreaker/             # 【本章新增】熔断
    │   ├── circuitbreaker.go       # 三态状态机 + 滑动窗口统计 + Group
    │   ├── circuitbreaker_test.go  # 熔断→半开→恢复闭环测试
    │   └── interceptor.go          # gRPC 客户端熔断拦截器
    ├── ratelimit/                  # 【本章新增】限流
    │   ├── tokenbucket.go          # 令牌桶（懒计算 + 按 key 维度）
    │   ├── slidingwindow.go        # 分桶滑动窗口
    │   └── ratelimit_test.go       # 两种限流器测试
    └── degradation/                # 【本章新增】降级
        └── degradation.go          # Strategy 接口 + Manager