7.1 配套代码：从延迟分解到优先级排序
对应小节：7.1 优先级金字塔 把「延迟分解的结果」自动映射到「该从哪一级开始优化」。
一、延迟分解 → 层级映射
# tools/prioritize.py
"""
从延迟分解结果推导优化优先级。
输入：各段的占比（可从第 6.2 节的 decompose-latency.py 得到）
输出：按层级排序的优化建议
"""
import sys
# 延迟分解的段 → 对应的金字塔层级与典型动作
SEGMENT_TO_LEVEL = {
"client_queue": {
"label": "网络 + 客户端排队",
"level": 1,
"level_name": "① 先问要不要做",
"actions": [
"减少返回字段（少传数据 → 省 CPU/内存/网络）",
"启用响应压缩（大响应体）",
"就近部署（CDN / 同可用区）",
"检查压测客户端是否与被测服务同机",
],
},
"service_queue": {
"label": "服务排队（进 handler 之前）",
"level": 3,
"level_name": "③ 架构与并发",
"actions": [
"查连接池 pending（第 6.5 节）",
"查线程池队列深度",
"检查 RUNNABLE 线程数 vs CPU 核数（调度器污染）",
"检查容器 CPU 节流（第 4.8 节）",
"检查是否有阻塞调用在 Dispatchers.Default 上",
],
},
"app_self": {
"label": "应用自身（序列化、业务逻辑）",
"level": 5,
"level_name": "⑤ 微观常数",
"actions": [
"采 CPU 火焰图找热点（第 6.3 节）",
"检查是否重复序列化（JSON → Map → JSON）",
"检查热路径日志",
"检查装箱/对象拷贝（收益通常是个位数 %）",
],
},
"postgres": {
"label": "PostgreSQL",
"level": 2,
"level_name": "② 算法与数据访问",
"actions": [
"查 pg_stat_statements 的 calls（N+1）与 mean（慢查询）",
"EXPLAIN (ANALYZE, BUFFERS) 检查执行计划",
"检查缺失的索引",
"检查事务范围是否过大",
],
},
"redis": {
"label": "Redis",
"level": 2,
"level_name": "② 算法与数据访问",
"actions": [
"检查大 key / 热 key",
"检查是否有 O(N) 命令（KEYS/HGETALL 大集合）",
"检查是否重复查询同一个 key",
],
},
"downstream": {
"label": "下游 HTTP",
"level": 3,
"level_name": "③ 架构与并发",
"actions": [
"检查是否串行调用多个独立下游（可并行化）",
"检查下游本身的健康度",
"加隔板（为每个下游分配独立资源）",
"加熔断（下游持续失败时快速失败）",
],
},
}
def prioritize(segments):
"""segments: [(key, 占比%)]，按占比排序并给出层级建议"""
print("═" * 78)
print("优化优先级：从延迟分解推导")
print("═" * 78)
print()
# 按占比排序
ranked = sorted(segments, key=lambda x: -x[1])
print(f"{'段':<30}{'占比':>8}{'金字塔层级':<20}{'优先级'}")
print("-" * 78)
for i, (key, pct) in enumerate(ranked, 1):
info = SEGMENT_TO_LEVEL.get(key)
if not info:
print(f"{key:<30}{pct:>7.1f}% {'（未知）':<20}")
continue
print(f"{info['label']:<30}{pct:>7.1f}% {info['level_name']:<20}第 {i}")
print()
print("═" * 78)
print("详细建议（按优先级）")
print("═" * 78)
for i, (key, pct) in enumerate(ranked, 1):
info = SEGMENT_TO_LEVEL.get(key)
if not info:
continue
print()
print(f"【优先级 {i}】{info['label']} —— {pct:.1f}%")
print(f" 层级：{info['level_name']}")
print(f" 优化收益上限：{pct:.1f}%（即使降为 0）")
print(f" 建议动作：")
for a in info["actions"]:
print(f" - {a}")
print()
print("═" * 78)
print("三个提醒：")
print(" ① 从占比最大的一段开始 —— 它的收益上限最高")
print(" ② 不要跳过「① 先问要不要做」—— 有时最好的优化是少做一件事")
print(" ③ 低级优化可能是高级优化的前提（例如先简化结构再加缓存），")
print(" 但这是「为了达成更大的目标」，不是为了顺手做")
print("═" * 78)
if __name__ == "__main__":
# 示例：从第 6.2 节的分解结果
example_segments = [
("client_queue", 5.0),
("service_queue", 72.0),
("app_self", 3.25),
("postgres", 16.25),
("redis", 1.0),
("downstream", 2.0),
]
prioritize(example_segments)
二、候选优化的排序器
# tools/rank-optimizations.py <optimizations.json>
"""
对候选优化排序：按「收益/成本比 + 金字塔层级」综合排序。
输入格式：
[
{
"name": "消除 N+1（批量查询）",
"level": 2,
"expected_benefit_pct": 70,
"cost": "低",
"risk": "低",
"effort_days": 0.5
},
...
]
"""
import json
import sys
LEVEL_WEIGHT = {1: 5.0, 2: 4.0, 3: 3.5, 4: 2.0, 5: 1.0, 6: 0.5}
COST_SCORE = {"低": 3.0, "中": 1.5, "高": 0.5}
RISK_SCORE = {"低": 3.0, "中": 1.5, "高": 0.5}
def score(opt):
"""
综合得分 = 预期收益 × 层级权重 × 成本分 × 风险分 / 投入天数^0.5
设计意图：
- 层级越高（数字越小）权重越大
- 成本/风险越高，得分越低
- 投入天数开根号（避免低估快速见效的优化）
"""
benefit = opt["expected_benefit_pct"]
lv = LEVEL_WEIGHT.get(opt["level"], 1.0)
cost = COST_SCORE.get(opt["cost"], 1.0)
risk = RISK_SCORE.get(opt["risk"], 1.0)
days = max(opt.get("effort_days", 1.0), 0.1)
return benefit * lv * cost * risk / (days ** 0.5)
def main(path):
opts = json.loads(open(path).read())
ranked = sorted(opts, key=score, reverse=True)
print("═" * 92)
print("候选优化排序（综合：收益 × 层级 × 成本 × 风险 / 投入）")
print("═" * 92)
print()
print(f"{'#':<4}{'优化':<34}{'层级':>5}{'收益%':>8}{'成本':>6}{'风险':>6}{'天数':>7}{'得分':>9}")
print("-" * 92)
for i, o in enumerate(ranked, 1):
print(f"{i:<4}{o['name'][:32]:<34}{o['level']:>5}{o['expected_benefit_pct']:>8.0f}"
f"{o['cost']:>6}{o['risk']:>6}{o.get('effort_days', 1):>7.1f}{score(o):>9.1f}")
print()
print("═" * 92)
print("建议的执行顺序（前 3 项）：")
for i, o in enumerate(ranked[:3], 1):
print(f" {i}. {o['name']}")
print(f" 层级 {o['level']}，预期收益 {o['expected_benefit_pct']}%，"
f"成本 {o['cost']}，风险 {o['risk']}，约 {o.get('effort_days', 1)} 天")
print()
print("注意：")
print(" ⚠️ 最后 2 项（通常是 JVM 调优与微观常数）建议【暂缓】——")
print(" 除非前面的优化已经做完且仍不达标")
print("═" * 92)
if __name__ == "__main__":
if len(sys.argv) < 2:
# 用示例数据演示
import tempfile, os
demo = [
{"name": "消除 N+1（批量查询）", "level": 2, "expected_benefit_pct": 70,
"cost": "低", "risk": "低", "effort_days": 0.5},
{"name": "把正则提到顶层", "level": 2, "expected_benefit_pct": 40,
"cost": "低", "risk": "低", "effort_days": 0.2},
{"name": "阻塞调用切到 Dispatchers.IO", "level": 3, "expected_benefit_pct": 60,
"cost": "低", "risk": "中", "effort_days": 0.5},
{"name": "加 Redis 缓存", "level": 4, "expected_benefit_pct": 50,
"cost": "中", "risk": "高", "effort_days": 3.0},
{"name": "换 ZGC", "level": 6, "expected_benefit_pct": 5,
"cost": "高", "risk": "高", "effort_days": 5.0},
{"name": "减少装箱", "level": 5, "expected_benefit_pct": 3,
"cost": "中", "risk": "低", "effort_days": 2.0},
]
with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as f:
json.dump(demo, f, ensure_ascii=False)
path = f.name
main(path)
os.unlink(path)
else:
main(sys.argv[1])
预期输出（示例数据）：
════════════════════════════════════════════════════════════════════════════════
候选优化排序（综合：收益 × 层级 × 成本 × 风险 / 投入）
════════════════════════════════════════════════════════════════════════════════
# 优化 层级 收益% 成本 风险 天数 得分
--------------------------------------------------------------------------------------------
1 把正则提到顶层 2 40 低 低 0.2 1341.6
2 消除 N+1（批量查询） 2 70 低 低 0.5 1187.9
3 阻塞调用切到 Dispatchers.IO 3 60 低 中 0.5 464.8
4 加 Redis 缓存 4 50 中 高 3.0 43.3
5 减少装箱 5 3 中 低 2.0 6.4
6 换 ZGC 6 5 高 高 5.0 1.7
建议的执行顺序（前 3 项）：
1. 把正则提到顶层
2. 消除 N+1（批量查询）
3. 阻塞调用切到 Dispatchers.IO
注意排序结果：「换 ZGC」排最后（得分 1.7），而「把正则提到顶层」排第一（1341.6）——差了 800 倍。这就是金字塔的量化体现。
三、动手改造
| 改动 | 观察什么 |
|---|---|
| 用你真实的延迟分解数据替换示例 | 得到你项目的优化优先级 |
把「加 Redis 缓存」的 risk 从高改成低 |
排名会上升——理解风险如何影响优先级 |
把「换 ZGC」的 expected_benefit_pct 改成 30 |
排名大幅上升——但这个数字需要有依据（不能为了让它排前面而改） |
给 LEVEL_WEIGHT 调整权重 |
体会「层级权重」如何主导排序 |
用 rank-optimizations.py 给你手上的优化任务排序 |
和直觉对比——差距在哪？ |
四、这段代码的局限
expected_benefit_pct是主观估计：它的准确性决定了排序的质量。应该用第 6 章的延迟分解来估（比如"数据库占 70%，所以消除 N+1 的上限是 70%"）。- 综合得分的公式是启发式的：权重（层级、成本、风险）可以按团队情况调整。
- 它不能替代判断：排序只是建议，实际还要考虑依赖关系（有些优化必须按顺序做）和业务优先级。
- 收益上限 ≠ 实际收益：比如"数据库占 70%“意味着消除 N+1 的上限是 70%，实际可能只降到 30%（因为还有连接开销等其他因素）。