服务熔断 · 降级 · 限流

Java // Resilience4j CircuitBreaker 配置与使用 import io.github.resilience4j.circuitbreaker.*; import java.time.Duration; // 1. 配置熔断器 CircuitBreakerConfig config = CircuitBreakerConfig.custom() .failureRateThreshold(50) // 失败率阈值 50% .slowCallRateThreshold(50) // 慢调用比例 50% .slowCallDurationThreshold(Duration.ofSeconds(3)) // 超过3秒算慢调用 .slidingWindowType(CircuitBreakerConfig.SlidingWindowType.COUNT_BASED) .slidingWindowSize(100) // 统计最近100次调用 .minimumNumberOfCalls(20) // 至少20次才判断 .waitDurationInOpenState(Duration.ofSeconds(10)) // OPEN→HALF_OPEN 冷却10s .permittedNumberOfCallsInHalfOpenState(5) // 半开时放行5个试探 .build(); // 2. 创建熔断器 CircuitBreaker circuitBreaker = CircuitBreaker.of("orderService", config); // 3. 包装调用 Supplier<String> decorated = CircuitBreaker .decorateSupplier(circuitBreaker, () -> remoteService.call()); // 4. 带降级的调用 try { String result = Try.ofSupplier(decorated) .recover(throwable -> "fallback: service unavailable") .get(); } catch (CallNotPermittedException e) { // 熔断打开时直接走到这里，不会真的调用 remoteService String result = "circuit breaker is OPEN"; }

Python # 手工实现一个简洁的熔断器（核心逻辑 < 50 行） import time import threading from enum import Enum class State(Enum): CLOSED = "closed" OPEN = "open" HALF_OPEN = "half_open" class CircuitBreaker: def __init__(self, fail_threshold=5, timeout=10, half_max=3): self.fail_threshold = fail_threshold self.timeout = timeout self.half_max = half_max self.state = State.CLOSED self.fail_count = 0 self.last_fail_time = 0 self.half_count = 0 self.lock = threading.Lock() def call(self, func, *args, **kwargs): with self.lock: if self.state == State.OPEN: if time.time() - self.last_fail_time > self.timeout: self.state = State.HALF_OPEN self.half_count = 0 else: raise Exception("CircuitBreaker is OPEN") if self.state == State.HALF_OPEN and self.half_count >= self.half_max: raise Exception("CircuitBreaker is OPEN (half-open limit)") try: result = func(*args, **kwargs) with self.lock: self.fail_count = 0 self.state = State.CLOSED return result except Exception: with self.lock: self.fail_count += 1 self.last_fail_time = time.time() if self.state == State.HALF_OPEN: self.half_count += 1 if self.fail_count >= self.fail_threshold: self.state = State.OPEN raise # 使用方式 cb = CircuitBreaker(fail_threshold=5, timeout=10) try: data = cb.call(remote_service.get_data, user_id=123) except Exception: data = "fallback: default response"

Go // Go 简洁版熔断器实现 package circuitbreaker import ( "errors" "sync" "time" ) type State int const ( StateClosed State = iota StateOpen StateHalfOpen ) type CircuitBreaker struct { mu sync.Mutex state State failureCount int failureThresh int timeout time.Duration lastFailTime time.Time halfOpenMax int halfOpenCount int } func New(failureThresh int, timeout time.Duration, halfMax int) *CircuitBreaker { return &CircuitBreaker{ state: StateClosed, failureThresh: failureThresh, timeout: timeout, halfOpenMax: halfMax, } } func (cb *CircuitBreaker) Call(fn func() error) error { cb.mu.Lock() if cb.state == StateOpen { if time.Since(cb.lastFailTime) > cb.timeout { cb.state = StateHalfOpen cb.halfOpenCount = 0 } else { cb.mu.Unlock() return errors.New("circuit breaker is OPEN") } } if cb.state == StateHalfOpen && cb.halfOpenCount >= cb.halfOpenMax { cb.mu.Unlock() return errors.New("half-open limit reached") } if cb.state == StateHalfOpen { cb.halfOpenCount++ } cb.mu.Unlock() err := fn() cb.mu.Lock() defer cb.mu.Unlock() if err != nil { cb.failureCount++ cb.lastFailTime = time.Now() if cb.failureCount >= cb.failureThresh { cb.state = StateOpen } return err } cb.failureCount = 0 cb.state = StateClosed return nil }

Java // Sentinel 流控规则 — 支持 QPS + 线程数两种模式 import com.alibaba.csp.sentinel.slots.block.flow.*; private void initFlowRules() { // QPS 限流：每秒最多 100 个请求 FlowRule rule = new FlowRule("createOrder") .setGrade(RuleConstant.FLOW_GRADE_QPS) // 基于 QPS .setCount(100) // QPS 阈值 .setControlBehavior(RuleConstant.CONTROL_BEHAVIOR_DEFAULT); FlowRuleManager.loadRules(Collections.singletonList(rule)); } // Sentinel 还支持： // - 线程数限流 (FLOW_GRADE_THREAD)：控制并发线程数 // - Warm Up 预热：系统冷启动时逐步放开阈值，避免被瞬间打垮 // - 匀速排队：漏桶模式，请求以均匀速度通过 // - 关联限流：当关联资源达到阈值，限流当前资源 // - 链路限流：只限制入口链路的流量 // 使用 @SentinelResource 注解 @SentinelResource( value = "createOrder", blockHandler = "createOrderBlockHandler" // 限流触发后执行 ) public Order createOrder(OrderRequest req) { return orderService.create(req); } public Order createOrderBlockHandler(OrderRequest req, BlockException e) { throw new BizException("系统繁忙，请稍后再试"); }

Nginx # Nginx 限流配置 — 基于漏桶算法 # 1. 定义限流区域（共享内存，10MB） # rate=10r/s 表示每秒10个请求 limit_req_zone $binary_remote_addr zone=mylimit:10m rate=10r/s; # 2. 在 server/location 中应用限流 server { listen 80; location /api/ { # burst=20: 允许突发最多排队20个请求（超出rate部分排队） # nodelay: 队列中的请求立即处理（不等待），但队列满则拒绝 limit_req zone=mylimit burst=20 nodelay; proxy_pass http://backend; } } # 3. 并发连接数限制 limit_conn_zone $binary_remote_addr zone=perip:10m; server { limit_conn perip 10; # 单个 IP 最多 10 个并发连接 } # 4. 下载速度限制 location /download/ { limit_rate 500k; # 限制下载速度 500KB/s limit_rate_after 10m; # 前10MB不限速 }

Python # 令牌桶算法手工实现 import time import threading class TokenBucket: def __init__(self, rate, capacity): self.rate = rate # 令牌生成速率 (个/秒) self.capacity = capacity # 桶容量（允许的最大突发量） self.tokens = capacity # 当前令牌数 self.last_time = time.time() self.lock = threading.Lock() def acquire(self, tokens=1): with self.lock: now = time.time() # 计算应该生成多少新令牌 elapsed = now - self.last_time new_tokens = elapsed * self.rate self.tokens = min(self.capacity, self.tokens + new_tokens) self.last_time = now if self.tokens >= tokens: self.tokens -= tokens return True # 放行 return False # 限流 # 使用：限制每秒100个请求，允许突发200个 bucket = TokenBucket(rate=100, capacity=200) async def rate_limit_middleware(request, call_next): if not bucket.acquire(): raise HTTPException(status_code=429, detail="Too Many Requests") return await call_next(request)