7.5 背压与韧性：保护自己也保护下游

本层的超时 < 上游给你的超时
且逐层递减

客户端 500ms
  └→ 网关 400ms
       └→ 应用 300ms
            ├→ 数据库 80ms
            ├→ 缓存 10ms
            └→ 下游服务 100ms

❌ 错误配置：应用给数据库的超时是 30 秒（默认值，没人改）

① 数据库变慢，每个查询要 3 秒
② 应用线程被占住 3 秒（因为超时是 30 秒，不会提前放弃）
③ 网关在 1 秒时超时返回 504 给用户
④ 但应用线程还在等数据库（还有 2 秒）
⑤ 新请求继续进来，占用新线程 → 线程池耗尽
⑥ 所有接口开始超时 → 服务雪崩
⑦ 流量回落后，那些连接还在等 → 【无法自愈】

// 统一的超时包装
class BudgetExceededException(message: String, cause: Throwable?) : RuntimeException(message, cause)

suspend fun <T> withBudget(ms: Long, name: String, block: suspend () -> T): T =
    try {
        withTimeout(ms) { block() }
    } catch (e: TimeoutCancellationException) {
        throw BudgetExceededException("$name exceeded ${ms}ms", e)
    }

// 使用：每层都有明确的预算
suspend fun getOrder(id: String): Order {
    val cached = withBudget(10, "redis") { cache.get(id) }
    if (cached != null) return cached

    return withBudget(80, "postgres") {
        repo.findById(id) ?: throw NotFoundException(id)
    }
}

// 检查超时链是否递减
data class TimeoutLayer(val name: String, val upstreamMs: Long, val selfMs: Long)

fun audit(layers: List<TimeoutLayer>): List<String> {
    val problems = mutableListOf<String>()
    layers.forEach { l ->
        if (l.selfMs > l.upstreamMs) {
            problems += "❌ ${l.name}: 本层超时 ${l.selfMs}ms > 上游给的 ${l.upstreamMs}ms"
        }
    }
    return problems
}

// ① 全局限流（保护自己）
val globalLimiter = RateLimiter.create(10_000.0)   // 每秒 1 万

// ② 按租户限流（防止单个租户占满）
val tenantLimiters = ConcurrentHashMap<String, RateLimiter>()
fun limiterFor(tenant: String) = tenantLimiters.computeIfAbsent(tenant) {
    RateLimiter.create(100.0)
}

// ③ 按接口限流（保护重接口）

关闭（正常）
  ↓ 错误率超过阈值
打开（快速失败，不调用下游）
  ↓ 等待一段时间
半开（放少量请求探测）
  ├─ 成功 → 回到关闭
  └─ 失败 → 回到打开

// ❌ 熔断器包得太外层——把参数校验失败也算作"下游故障"
circuitBreaker.execute {
    validateParams(params)      // 参数错误不应该触发熔断！
    callDownstream()
}

// ✅ 只对真正的下游调用熔断
validateParams(params)
circuitBreaker.execute { callDownstream() }

服务 A 调用下游 X 和 Y
X 变慢 → A 的线程池被 X 的调用占满
→ A 无法处理任何请求（包括不需要 X 的）
→ Y 也被牵连

// ① 每个下游独立的信号量
val xLimiter = Semaphore(32)
val yLimiter = Semaphore(32)

suspend fun callX() = xLimiter.withPermit { httpGetX() }
suspend fun callY() = yLimiter.withPermit { httpGetY() }

// ② 或者每个下游独立的线程池 / 调度器
private val xDispatcher = Dispatchers.IO.limitedParallelism(32)
private val yDispatcher = Dispatchers.IO.limitedParallelism(32)

// ③ 数据库侧：读写分离、不同库用不同连接池

// ❌ 无约束的重试：放大故障
repeat(3) { callDownstream() }

// ✅ 有约束的重试
suspend fun <T> retryWithBackoff(
    maxAttempts: Int = 3,
    initialDelayMs: Long = 100,
    maxDelayMs: Long = 2_000,
    block: suspend (attempt: Int) -> T,
): T {
    var lastException: Throwable? = null
    repeat(maxAttempts) { attempt ->
        try {
            return block(attempt)
        } catch (e: RetryableException) {          // ← 只重试可重试的异常
            lastException = e
            if (attempt < maxAttempts - 1) {
                // 指数退避 + 随机抖动
                val base = minOf(initialDelayMs shl attempt, maxDelayMs)
                val jitter = Random.nextLong(0, base / 2)
                delay(base + jitter)
            }
        }
    }
    throw lastException!!
}

如果每个请求最多重试 2 次，那么最坏情况下：
  下游承受的流量 = 正常流量 × 3

所以在限流和容量规划时，要按 ×3 算（或限制重试的总比例）

class DownstreamClient(
    private val http: HttpClient,
    private val limiter: Semaphore,                  // 隔板
    private val rateLimiter: RateLimiter,            // 限流
    private val circuitBreaker: CircuitBreaker,      // 熔断
) {
    suspend fun fetch(id: String): Result<Data> {
        // ① 限流（自己这一侧的速率控制）
        if (!rateLimiter.tryAcquire()) {
            return Result.error(BusinessException("请求过于频繁"))
        }

        // ② 熔断（快速失败，不调用下游）
        if (circuitBreaker.isOpen()) {
            return Result.error(BusinessException("下游暂时不可用"))
        }

        return try {
            // ③ 隔板 + 超时 + 重试
            retryWithBackoff(maxAttempts = 2) {
                limiter.withPermit {
                    withBudget(100, "downstream") {
                        http.get("$BASE/$id")
                    }
                }
            }.let {
                circuitBreaker.recordSuccess()
                Result.success(it)
            }
        } catch (e: BudgetExceededException) {
            circuitBreaker.recordFailure()
            Result.error(RemoteException("下游超时", e))
        } catch (e: Exception) {
            circuitBreaker.recordFailure()
            Result.error(RemoteException("下游失败", e))
        }
    }
}