1.5 配套代码：延迟预算的自洽性检查与超时链推导
对应小节：1.5 延迟预算 两份代码解决两个问题：① 预算表加起来超了没有？② 超时设置是不是逐层递减的？
一、预算表的数据结构
// src/main/kotlin/budget/LatencyBudget.kt
package budget
/**
* 一个预算条目。
* @param parallelGroup 非 null 表示"同组内的条目是并行执行的"（取最大值而非相加）
*/
data class BudgetEntry(
val name: String,
val p99Ms: Double,
val parallelGroup: String? = null,
val note: String = "",
)
class LatencyBudget(
val sloP99Ms: Double,
val reserveMs: Double, // 余量（必须 > 0）
val entries: List<BudgetEntry>,
) {
/** 串行段之和 + 每个并行组的最大值 + 余量 */
fun totalP99Ms(): Double {
val serial = entries.filter { it.parallelGroup == null }.sumOf { it.p99Ms }
val parallel = entries.filter { it.parallelGroup != null }
.groupBy { it.parallelGroup!! }
.map { (_, group) -> group.maxOf { it.p99Ms } }
.sum()
return serial + parallel + reserveMs
}
fun validate(): List<String> {
val problems = mutableListOf<String>()
val total = totalP99Ms()
if (reserveMs <= 0) {
problems += "余量必须为正数：没有余量的预算在流量波动时必然崩（见第 0 章 0.5 节）"
}
if (total > sloP99Ms) {
problems += "预算合计 %.0fms 超过 SLO %.0fms，超出 %.0fms —— 必须砍掉某一跳或放宽 SLO"
.format(total, sloP99Ms, total - sloP99Ms)
}
if (entries.none { it.name.contains("排队") }) {
problems += "警告：预算表里没有「排队」这一项。请求在队列里等待的时间也必须计入预算"
}
val reserveRatio = reserveMs / sloP99Ms
if (reserveRatio in 0.0001..0.05) {
problems += "余量只占 %.1f%%，偏少（建议 10%%~20%%）".format(reserveRatio * 100)
}
return problems
}
fun report(): String = buildString {
appendLine("延迟预算（SLO = %.0fms，余量 = %.0fms）".format(sloP99Ms, reserveMs))
appendLine("-".repeat(64))
val serial = entries.filter { it.parallelGroup == null }
val parallelGroups = entries.filter { it.parallelGroup != null }.groupBy { it.parallelGroup!! }
serial.forEach { appendLine(" %-32s %8.1fms".format(it.name, it.p99Ms)) }
parallelGroups.forEach { (g, list) ->
appendLine(" [并行组 $g]")
list.forEach { appendLine(" %-30s %8.1fms".format(it.name, it.p99Ms)) }
appendLine(" %-30s %8.1fms ← 并行取最大值".format("小计", list.maxOf { it.p99Ms }))
}
appendLine(" %-32s %8.1fms".format("余量", reserveMs))
appendLine("-".repeat(64))
appendLine(" %-32s %8.1fms".format("合计", totalP99Ms()))
appendLine(" %-32s %8.1fms".format("SLO", sloP99Ms))
appendLine()
val problems = validate()
if (problems.isEmpty()) {
appendLine("✅ 校验通过")
} else {
problems.forEach { appendLine("❌ $it") }
}
}
}
fun main() {
// ── 反例：故意算错的预算（正文里的例子） ──────────────────
val bad = LatencyBudget(
sloP99Ms = 200.0, reserveMs = 0.0,
entries = listOf(
BudgetEntry("客户端 → 网关", 25.0),
BudgetEntry("网关处理", 15.0),
BudgetEntry("应用排队", 20.0),
BudgetEntry("应用逻辑", 50.0),
BudgetEntry("PostgreSQL", 60.0),
BudgetEntry("Redis", 5.0),
BudgetEntry("下游 HTTP", 40.0),
BudgetEntry("响应传输", 10.0),
),
)
println("=== 反例：忘了做加法 ===")
println(bad.report())
println()
// ── 正例：压缩后并留出余量 ────────────────────────────────
val good = LatencyBudget(
sloP99Ms = 200.0, reserveMs = 20.0,
entries = listOf(
BudgetEntry("客户端 → 网关", 25.0),
BudgetEntry("网关处理", 15.0),
BudgetEntry("应用排队", 20.0),
BudgetEntry("应用逻辑", 45.0),
BudgetEntry("PostgreSQL", 40.0),
BudgetEntry("Redis", 5.0),
BudgetEntry("下游 HTTP", 25.0),
BudgetEntry("响应传输", 5.0),
),
)
println("=== 正例：压缩 + 留余量 ===")
println(good.report())
println()
// ── 并行调用的写法 ────────────────────────────────────────
val withParallel = LatencyBudget(
sloP99Ms = 200.0, reserveMs = 20.0,
entries = listOf(
BudgetEntry("客户端 → 网关", 25.0),
BudgetEntry("网关处理", 15.0),
BudgetEntry("应用排队", 20.0),
BudgetEntry("用户服务", 60.0, parallelGroup = "fanout"),
BudgetEntry("商品服务", 80.0, parallelGroup = "fanout"),
BudgetEntry("库存服务", 45.0, parallelGroup = "fanout"),
),
)
println("=== 并行扇出：取最大值，不是相加 ===")
println(withParallel.report())
println("说明：如果按相加算 = 185ms（早就超了），实际只需 80ms（最慢的那个）。")
}
二、预期输出
=== 反例：忘了做加法 ===
延迟预算（SLO = 200ms，余量 = 0ms）
----------------------------------------------------------------
客户端 → 网关 25.0ms
网关处理 15.0ms
应用排队 20.0ms
应用逻辑 50.0ms
PostgreSQL 60.0ms
Redis 5.0ms
下游 HTTP 40.0ms
响应传输 10.0ms
余量 0.0ms
----------------------------------------------------------------
合计 225.0ms
SLO 200.0ms
❌ 预算合计 225ms 超过 SLO 200ms，超出 25ms —— 必须砍掉某一跳或放宽 SLO
❌ 余量必须为正数：没有余量的预算在流量波动时必然崩
三、超时链推导（更救命的第二份代码）
规则只有一条：本层超时必须小于上游给的时间，且逐层递减。
// src/main/kotlin/budget/TimeoutChain.kt
package budget
data class TimeoutNode(
val layer: String,
val upstreamBudgetMs: Double, // 上游给这一层的时间
val selfOverheadMs: Double, // 这一层自身的处理开销（不含下游）
val target: String = "", // 这一层要调用的下游
)
data class TimeoutPlan(
val layer: String,
val selfTimeoutMs: Double,
val remainingForDownstreamMs: Double,
val problems: List<String>,
)
/**
* 推导超时链。
*
* @param retryBudgetRatio 如果要重试，需要为"多次尝试 + 退避"预留时间（例如 0.5 表示留一半）
*/
fun deriveTimeouts(nodes: List<TimeoutNode>, retryBudgetRatio: Double = 0.0): List<TimeoutPlan> {
var upstream = Double.MAX_VALUE
return nodes.map { node ->
val problems = mutableListOf<String>()
// ① 本层超时必须小于上游给的预算
val maxSelf = if (upstream == Double.MAX_VALUE) node.upstreamBudgetMs
else minOf(node.upstreamBudgetMs, upstream)
var selfTimeout = maxSelf - node.selfOverheadMs
if (selfTimeout <= 0) {
problems += "上游只给了 ${maxSelf}ms，但本层自身开销就要 ${node.selfOverheadMs}ms —— 必然超时"
selfTimeout = 1.0
}
// ② 如果要重试，单次超时必须更短
if (retryBudgetRatio > 0) {
selfTimeout *= (1 - retryBudgetRatio)
}
// ③ 留给下游的时间必须更少
val remaining = selfTimeout - node.selfOverheadMs
if (remaining <= 0) {
problems += "扣掉自身开销后没有时间留给下游「${node.target}」，下游必须降级或异步化"
}
upstream = remaining
TimeoutPlan(
layer = node.layer,
selfTimeoutMs = selfTimeout,
remainingForDownstreamMs = maxOf(remaining, 0.0),
problems = problems,
)
}
}
fun main() {
println("=== 反例：下游超时比上游还长（级联雪崩的配方）===")
val badPlan = deriveTimeouts(
listOf(
TimeoutNode("客户端 → 网关", 300.0, 5.0),
TimeoutNode("网关", 250.0, 15.0, "应用"),
TimeoutNode("应用（给 PostgreSQL）", 5000.0, 10.0), // ← 默认 5 秒，没人改过
)
)
badPlan.forEach { println(" ${it.layer}: 自身超时 %.0fms".format(it.selfTimeoutMs)) }
println(" ⚠️ 应用给 PostgreSQL 的超时是 5 秒，而网关只给应用 250ms ——")
println(" 数据库一慢，应用线程会被占住 5 秒，而网关早已放弃。线程耗尽 → 雪崩。")
println()
println("=== 正例：逐层递减 + 重试预算 ===")
val goodPlan = deriveTimeouts(
listOf(
TimeoutNode("客户端 → 网关", 300.0, 5.0),
TimeoutNode("网关", 250.0, 15.0, "应用"),
TimeoutNode("应用", 200.0, 20.0, "PostgreSQL/Redis/下游"),
TimeoutNode("应用 → PostgreSQL", 120.0, 5.0),
),
retryBudgetRatio = 0.0,
)
goodPlan.forEach {
println(" %-24s 自身超时 %6.0fms，留给下游 %6.0fms".format(
it.layer, it.selfTimeoutMs, it.remainingForDownstreamMs))
it.problems.forEach { p -> println(" ❌ $p") }
}
println()
println("=== 带重试的情况：单次超时必须更短 ===")
deriveTimeouts(
listOf(
TimeoutNode("应用（给下游，允许重试 2 次）", 200.0, 20.0, "下游服务"),
),
retryBudgetRatio = 0.6, // 留 60% 给重试与退避
).forEach {
println(" %-24s 单次超时 %6.0fms（重试 2 次的总时间预算另算）".format(it.layer, it.selfTimeoutMs))
}
}
输出要点：
=== 正例：逐层递减 + 重试预算 ===
客户端 → 网关 自身超时 295ms，留给下游 290ms
网关 自身超时 235ms，留给下游 220ms
应用 自身超时 200ms，留给下游 180ms
应用 → PostgreSQL 自身超时 115ms，留给下游 110ms
每一层的超时都比上层小，这就是正确的超时链。
四、把两份代码接进日常
| 时机 | 做什么 |
|---|---|
| 新增接口时 | 写一份 LatencyBudget，跑 validate()，超了就重新分配 |
| 配置超时时 | 用 deriveTimeouts() 推导，禁止手写「默认 30 秒」 |
| 发布前 | 把预算表的「实测」列填满（第 4 章有了观测之后），对比是否超预算 |
| 线上告警时 | 看哪一跳超预算，直接定位 |
五、动手改造
| 改动 | 观察什么 |
|---|---|
把 reserveMs 改成 0.5 |
校验会提示余量偏少——想想 0.5ms 的余量能吸收什么抖动 |
| 给「应用逻辑」加入并行组 | 检查合计值的变化（相加 → 取最大） |
在 deriveTimeouts 里把最后一跳的 selfOverheadMs 调到超过上游预算 |
观察「必然超时」的报错 |
把 retryBudgetRatio 从 0 改到 0.8 |
单次超时被压缩到什么程度？重试还有意义吗？ |
六、这段代码的局限
- 预算表用的是 P99，但 P99 不可加（第 1.2 节）。这份代码把各跳 P99 相加当作目标上限来校验自洽性是合理的，但不能用它推算真实的总 P99——真实值必须实测。
- 并行组的建模比较粗糙（只区分「并行」和「串行」）。真实链路里常有嵌套的并行/串行混合结构，建议用 tracing 的 span 树来表达。
- 它不知道「重试会放大下游压力」。重试预算只算了时间，没算流量放大——后者要在第 7 节的背压设计里处理。