9.7 配套代码：优化、验证与否决

#!/usr/bin/env python3
"""tools/step6a-plan.py —— 按金字塔排序优化项

第 7.1 节的优先级金字塔：
  ① 去掉不必要的工作（最大收益、最小风险）
  ② 减少工作量的算法/数据结构改进
  ③ 减少等待（并发/池/批量化）
  ④ 让剩下的更快（微优化、JIT 调参）
  ⑤ 加资源（最后一招）

本 Lab 的排序（算出「预期收益 / 风险」后得出）：
"""
import json
import pathlib

PLAN = [
    {
        "order": 1,
        "name": "给 code 加唯一索引",
        "layer": "① 去掉不必要的工作",
        "why": "Seq Scan 扫描 100 万行只为找 1 行 —— 99.9999% 是纯粹浪费",
        "expected": "P99 420.5ms → 45ms（依赖级 380ms → ~2ms）",
        "risk": "极低（一条 DDL；CONCURRENTLY 建索引不锁表）",
        "cost": "5 分钟",
        "single_variable": "只加索引，其他一律不动",
    },
    {
        "order": 2,
        "name": "阻塞调用切到 Dispatchers.IO + Semaphore 限流",
        "layer": "③ 减少等待（隔离）",
        "why": "JDBC 阻塞占满 Default 调度器（并行度=核数），"
               "导致不查库的 /health 也慢",
        "expected": "/health P99 380ms → 8ms（其他接口也跟着稳）",
        "risk": "低（但需限流，否则 IO 池会被打爆）",
        "cost": "30 分钟",
        "single_variable": "只切调度器 + 限流，不碰索引/日志",
    },
    {
        "order": 3,
        "name": "日志异步化 + 降级 + 参数化",
        "layer": "① 去掉不必要的工作",
        "why": "热路径 INFO 日志（含完整 UA/IP）同步写盘，"
               "wall 中占 6.1%",
        "expected": "P99 改善约 5%（45ms → 42.7ms）",
        "risk": "极低（异步 appender + 去掉 UA/IP）",
        "cost": "20 分钟",
        "single_variable": "只改日志",
    },
    {
        "order": 4,
        "name": "synchronized → Mutex（或去掉锁）",
        "layer": "② 减少工作量 / ③ 减少等待",
        "why": "线程级锁在协程里会跨越挂起点；写路径偶发 BLOCKED",
        "expected": "写接口 P99 改善约 1.2%",
        "risk": "中（改锁容易引入并发 bug，需要压测验证）",
        "cost": "1 小时",
        "single_variable": "只改锁",
    },
    {
        "order": 5,
        "name": "缓存 TTL 加抖动（±20%）",
        "layer": "③ 减少等待（削峰）",
        "why": "固定 TTL 导致周期性缓存雪崩（埋雷 ③）",
        "expected": "消除浸泡时的 DB QPS 脉冲（P99 未必改善，"
                    "但尖刺消失）",
        "risk": "极低",
        "cost": "10 分钟",
        "single_variable": "只改 TTL 策略",
    },
]

REJECTED = [
    {
        "name": "异步批量 UPDATE hits",
        "layer": "③ 减少等待（批量化）",
        "measured_gain": "2.1%",
        "mdd": "11.8%",
        "reason": "2.1% < MDD 11.8% —— 统计上无法区分于噪声",
        "extra_cost": "引入最终一致性：hits 会滞后；"
                      "需处理进程崩溃时的计数丢失；需额外的批量队列",
        "decision": "❌ 否决",
        "revisit_when": "当 hits 写入成为 top1 瓶颈时（当前排第 3）",
    },
    {
        "name": "加 Redis 做二级缓存",
        "layer": "① 去掉不必要的工作",
        "measured_gain": "未测（估算 5%~10%）",
        "mdd": "11.8%",
        "reason": "本地 Caffeine 命中率已达 75%，"
                  "剩余 25% 中大部分是【长尾冷 key】（幂律分布）——"
                  "Redis 也缓存不住",
        "extra_cost": "新增一个依赖（可用性风险）；"
                      "网络往返约 0.5ms；运维成本；"
                      "缓存一致性复杂度",
        "decision": "❌ 否决",
        "revisit_when": "当需要多实例共享缓存（水平扩展）时",
    },
]


def main(out: str) -> int:
    d = pathlib.Path(out)
    d.mkdir(parents=True, exist_ok=True)

    print("═══ 优化计划（按金字塔排序）═══\n")
    for p in PLAN:
        print(f"{p['order']}. {p['name']}")
        print(f"   层次：{p['layer']}")
        print(f"   理由：{p['why']}")
        print(f"   预期：{p['expected']}")
        print(f"   风险：{p['risk']}　成本：{p['cost']}")
        print(f"   ⚠️  单变量：{p['single_variable']}")
        print()

    print("═══ 被否决的方案 ═══\n")
    for r in REJECTED:
        print(f"{r['decision']}  {r['name']}")
        print(f"   层次：{r['layer']}")
        print(f"   实测收益：{r['measured_gain']}　MDD：{r['mdd']}")
        print(f"   否决理由：{r['reason']}")
        print(f"   额外代价：{r['extra_cost']}")
        print(f"   重新评估条件：{r['revisit_when']}")
        print()

    # ── 关键纪律检查 ──
    print("═══ 纪律检查 ═══")
    print()
    print("✅ 每一项都写了「单变量」约束")
    print("✅ 被否决的方案都记录了理由与重新评估条件")
    print("✅ 排序依据是「金字塔层次」，不是「我觉得哪个好改」")
    print()
    print("⚠️  没有记录「为什么不做 X」的报告 = 看起来像没考虑过备选")

    (d / "optimization-plan.json").write_text(json.dumps(
        {"plan": PLAN, "rejected": REJECTED}, indent=2, ensure_ascii=False))
    print(f"\n✅ 计划 → {d}/optimization-plan.json")
    return 0


if __name__ == "__main__":
    import sys
    sys.exit(main(sys.argv[1] if len(sys.argv) > 1
                  else "perf/results/step6-optimize"))

-- perf/sql/02-add-code-index.sql
-- ═══ 优化 1：把 Seq Scan 变成 Index Scan ═══

-- 【关键】用 CONCURRENTLY 避免锁表
-- 普通 CREATE INDEX 会持有 SHARE 锁，阻塞所有写操作
-- CONCURRENTLY 的代价：更慢（要扫两遍表），但线上安全
CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS idx_links_code
    ON links (code);

-- 建完后必须 ANALYZE，否则 Planner 可能仍选 Seq Scan
ANALYZE links;

-- ── 验证 ──
-- 预期：Index Scan / Index Only Scan，Execution Time < 1ms
EXPLAIN (ANALYZE, BUFFERS)
SELECT id, code, url, user_id, created_at, hits
FROM links WHERE code = 'aB3xK9';

 Index Scan using idx_links_code on links  (cost=0.42..8.44 rows=1 width=48)
                                            (actual time=0.028..0.030 rows=1 loops=1)
   Index Cond: ((code)::text = 'aB3xK9'::text)
   Buffers: shared hit=4
 Planning Time: 0.112 ms
 Execution Time: 0.031 ms

// repository/LinkRepository.kt（改动后）
package shortlink.repository

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.sync.Semaphore
import kotlinx.coroutines.sync.withPermit
import kotlinx.coroutines.withContext

class LinkRepository(private val ds: DataSource) {

    /**
     * ═══ 优化 2：把阻塞的 JDBC 调用搬到 Dispatchers.IO ═══
     *
     * ⚠️ 关键：光切调度器是不够的！
     *    Dispatchers.IO 有 64 个线程（默认），如果不限流，
     *    高并发时会瞬间起 64 个线程全部阻塞在 DB 上，
     *    把数据库打爆（第 7.6 节）。
     *
     *    所以必须加 Semaphore 限流，让"想查库的协程"排队而不是
     *    "无限地去占线程"。
     */
    private val dbLimiter = Semaphore(permits = 32)

    suspend fun findByCode(code: String): Link? =
        dbLimiter.withPermit {
            withContext(Dispatchers.IO) {
                ds.connection.use { c ->
                    c.prepareStatement(
                        "SELECT id, code, url, user_id, created_at, hits " +
                        "FROM links WHERE code = ?"
                    ).use { ps ->
                        ps.setString(1, code)
                        ps.executeQuery().use { rs ->
                            if (rs.next()) rs.toLink() else null
                        }
                    }
                }
            }
        }

    /**
     * 为什么 Semaphore(32) 而不是 64？
     *   32 是【与数据库协商的结果】——DB max_connections=200，
     *   单实例占 32，留足余量给其他实例与运维工具。
     *
     *   为什么不用 HikariCP 的 maximumPoolSize 来限流？
     *   Hikari 的等待是【阻塞线程】的（threadsAwaitingConnection），
     *   而 Semaphore 让协程【挂起】（不占线程）——
     *   后者在高并发下更省资源。
     */
}

因为优化 1 已经把 DB 查询从 380ms 压到 0.03ms ——
此时跳转的耗时大头已经不是"等 DB"，而是别的（序列化、日志等）。
所以优化 2 对跳转的边际收益小。

但优化 2 消除了一个【系统性风险】：
  不查库的接口会被查库的接口拖垮。
  这个风险在"DB 突然变慢时"会放大：
  DB 一慢 → Default 被占满 → 整个服务（含健康检查）不可用
  → K8s 健康检查失败 → 实例被重启 → 雪崩

修复它不需要它带来 P99 提升 —— 它带来的是【可用性】。
这批注必须写进报告，否则会被质疑"改了没用"。

<!-- src/main/resources/logback.xml -->
<configuration>
  <!-- ═══ 优化 3：异步 appender ═══ -->
  <!-- 关键：neverBlock=true —— 队列满时【丢弃】日志而不是阻塞业务 -->
  <appender name="ASYNC_FILE" class="ch.qos.logback.classic.AsyncAppender">
    <appender-ref ref="FILE" />
    <queueSize>8192</queueSize>
    <discardingThreshold>0</discardingThreshold>
    <neverBlock>true</neverBlock>
    <includeCallerData>false</includeCallerData>
  </appender>

  <appender name="FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
    <file>logs/app.log</file>
    <encoder>
      <!-- 简化格式：去掉了线程名、logger 名（热路径不需要） -->
      <pattern>%d{HH:mm:ss.SSS} %-5level %msg%n</pattern>
    </encoder>
    <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
      <fileNamePattern>logs/app.%d{yyyy-MM-dd}.%i.log.gz</fileNamePattern>
      <maxFileSize>100MB</maxFileSize>
      <maxHistory>7</maxHistory>
    </rollingPolicy>
    <immediateFlush>false</immediateFlush>   <!-- 关键：不要每条 flush -->
  </appender>

  <!-- 关键：把热路径的 logger 降到 WARN -->
  <logger name="shortlink.http" level="WARN" />
  <logger name="io.ktor" level="WARN" />

  <root level="INFO">
    <appender-ref ref="ASYNC_FILE" />
  </root>
</configuration>

// http/Routes.kt（改动后）—— 热路径日志的正确写法
get("/{code}") {
    val code = call.parameters["code"]!!

    // ═══ 优化 3：热路径日志的正确形态 ═══
    //
    // 原则：
    //  ① 热路径默认不打日志（用指标代替 —— 指标是内存计数，几乎免费）
    //  ② 必须打时用【参数化】（不拼接字符串，级别关掉时零成本）
    //  ③ 不打大字段（完整 URL / UA / IP 对排障价值低但开销高）
    //  ④ 需要时用 debug，采样打
    //
    // 注意 ② 的含义：
    //   log.info("code={}", code)   ← 级别为 WARN 时，不构造字符串，零成本
    //   log.info("code=" + code)    ← 无论如何都拼接字符串，有成本
    log.debug("redirect code={}", code)

    val link = service.resolve(code)
    ...
}

后者在【调用 log.info 之前】就已经拼好了字符串 ——
即使日志级别是 WARN（不会输出），字符串也已经分配了。

前者把 code 作为参数传给 logger，logger 先判断级别：
  WARN 级别 → 直接 return，【不构造字符串】
  DEBUG 级别 → 才做格式化

在高频调用下，这个差别就是"每秒分配几十万个临时字符串"
vs "什么都不分配"。

// repository/LinkCache.kt（改动后）
package shortlink.repository

import com.github.benmanes.caffeine.cache.Cache
import com.github.benmanes.caffeine.cache.Caffeine
import com.github.benmanes.caffeine.cache.Expiry
import shortlink.model.Link
import java.time.Duration
import java.util.concurrent.ThreadLocalRandom

class LinkCache {

    /**
     * ═══ 优化 4：TTL 加抖动 ═══
     *
     * 问题：所有 key 固定 10 分钟 TTL
     *   → 同一批写入的 key 在【同一时刻】集体失效
     *   → DB QPS 出现周期性脉冲
     *
     * 修复：每个 key 的 TTL 在 [8min, 12min] 之间随机
     *   → 失效时刻被打散到 4 分钟窗口内
     *   → 脉冲变成平缓的波浪
     *
     * 注意：抖动只在【写入时】算一次，同一个 key 的 TTL 是固定的。
     *      不要在每次读取时重新随机 —— 那会导致 TTL 无限延长。
     */
    private val baseTtl = Duration.ofMinutes(10)
    private val jitterRatio = 0.2

    private val cache: Cache<String, Link> = Caffeine.newBuilder()
        .maximumSize(100_000)
        .expireAfter(Expiry { _, link, _ ->
            val base = baseTtl.toNanos()
            val jitter = (base * jitterRatio).toLong()
            val ttl = base + ThreadLocalRandom.current().nextLong(-jitter, jitter)
            // ── 返回值单位是纳秒 ──
            ttl
        })
        .recordStats()
        .build()

    fun get(code: String): Link? = cache.getIfPresent(code)

    fun put(code: String, link: Link) {
        cache.put(code, link)
    }

    fun invalidate(code: String) {
        cache.invalidate(code)
    }

    fun hitRate(): Double = cache.stats().hitRate()
    fun stats() = cache.stats()
}

#!/usr/bin/env bash
# tools/step6b-verify.sh —— 对每个优化执行「A/B/A'」验证
#
# ═══ 三问纪律（第 7.8 节）═══
#   ① 提升多少？（必须有对照组，且 >= MDD）
#   ② 代价是什么？（复杂度、资源、新的风险）
#   ③ 回归验证了吗？（A' 回滚验证 + 回归测试）
set -euo pipefail

OPT="${1:?用法: $0 <优化编号 1-5>}"
OUT="perf/results/step6-optimize/opt-$OPT"
mkdir -p "$OUT"

echo "═══ 优化 $OPT 的 A/B 验证 ═══"
echo

run_phase() {
  local phase="$1"     # A / B / Aprime
  echo "─── 阶段 $phase ───"
  echo "静置 30s..."
  sleep 30

  curl -sf http://localhost:8080/metrics > "$OUT/metrics-$phase.txt" || true

  (cd perf/k6 && k6 run \
    --summary-export="$OLDPWD/$OUT/summary-$phase.json" \
    --quiet baseline-single.js 2>&1 | tail -20) \
    || echo "  （k6 退出码非 0，检查是否有阈值）"

  python3 - "$OUT/summary-$phase.json" "$phase" <<'PY'
import json, sys
p, phase = sys.argv[1], sys.argv[2]
s = json.loads(open(p).read())["metrics"]
m = s["http_req_duration"]
print(f"  {phase}: P50={m['p(50)']:.1f}  P95={m['p(95)']:.1f}  "
      f"P99={m['p(99)']:.1f}  RPS={s['http_reqs']['rate']:.1f}")
PY
  echo
}

echo "① 阶段 A：优化前（基线）"
run_phase A
echo "   → 请现在【应用】优化 $OPT，然后按回车"
read -r

echo "② 阶段 B：优化后"
run_phase B

echo "③ 阶段 A'：回滚验证"
echo "   → 请现在【回滚】优化 $OPT，然后按回车"
read -r
run_phase Aprime

# ── 汇总判定 ──
python3 - "$OUT" <<'PY'
import json, pathlib, sys, statistics

d = pathlib.Path(sys.argv[1])

def p99(phase):
    s = json.loads((d / f"summary-{phase}.json").read_text())["metrics"]
    return s["http_req_duration"]["p(99)"]

a, b, ap = p99("A"), p99("B"), p99("Aprime")

# 读噪声底线
nf = pathlib.Path("perf/results/step3-baseline/noise-floor.json")
if nf.exists():
    mdd = json.loads(nf.read_text())["mdd_pct"]
else:
    mdd = 11.8
    print("⚠️  未找到噪声底线，用默认 MDD=11.8%")

delta = (b - a) / a * 100

print()
print("═══ 三问判定 ═══")
print()
print(f"① 提升多少？")
print(f"   A  (优化前)  {a:.1f} ms")
print(f"   B  (优化后)  {b:.1f} ms")
print(f"   变化         {delta:+.1f}%")
print(f"   MDD          ±{mdd:.1f}%")
if abs(delta) >= mdd:
    verdict = "显著改善 ✅" if delta < 0 else "显著变差 ❌"
    print(f"   判定         {verdict}")
else:
    print(f"   判定         在噪声内 —— 【无法确认效果】⚠️")
    print(f"   含义         不是「没效果」，是「这次测不出来」")
    print(f"                要确认需增加轮次或改用更敏感的指标")

print()
print(f"② 代价是什么？")
print(f"   → 需人工填写：复杂度 / 资源 / 新风险")
print(f"      （脚本无法判断，这是评审时的必答项）")

print()
print(f"③ 回归验证？")
print(f"   A' (回滚后)  {ap:.1f} ms")
drift = abs(ap - a) / a * 100
print(f"   与 A 的偏差   {drift:.1f}%")
if drift <= mdd:
    print(f"   判定         回到基线 ✅（证明是优化本身的效果，不是环境漂移）")
else:
    print(f"   判定         未回到基线 ❌")
    print(f"   含义         环境在漂移，或优化有【残留副作用】")
    print(f"                这次对比结论【不可信】，需重做")
PY

═══ 三问判定 ═══

① 提升多少？
   A  (优化前)  420.5 ms
   B  (优化后)  45.2 ms
   变化         -89.3%
   MDD          ±11.8%
   判定         显著改善 ✅

② 代价是什么？
   → 需人工填写：复杂度 / 资源 / 新风险

③ 回归验证？
   A' (回滚后)  418.9 ms
   与 A 的偏差   0.4%
   判定         回到基线 ✅（证明是优化本身的效果，不是环境漂移）

═══ 三问判定 ═══

① 提升多少？
   A  (优化前)  45.2 ms
   B  (优化后)  44.3 ms
   变化         -2.1%
   MDD          ±11.8%
   判定         在噪声内 —— 【无法确认效果】⚠️
   含义         不是「没效果」，是「这次测不出来」
                要确认需增加轮次或改用更敏感的指标

② 代价是什么？
   → 引入最终一致性；hits 会滞后
   → 需处理进程崩溃时的计数丢失
   → 需要额外的批量队列与定时刷写

③ 回归验证？
   A' (回滚后)  45.5 ms
   与 A 的偏差   0.7%
   判定         回到基线 ✅

收益 2.1%（测不出来）+ 明显代价（一致性 + 复杂度）
→ **否决**

这个否决算【结论】，不是【失败】——
它证明了你评估过并且有数据支撑。