文档目录

6.9 配套代码：Lab 6 编排与证据矩阵

对应小节：6.9 Lab 6 三个埋雷接口 + 分批取证 + 证据矩阵自动生成。

一、三个埋雷接口

// src/main/kotlin/lab6/BottleneckApp.kt
package lab6

import io.ktor.server.application.*
import io.ktor.server.response.*
import io.ktor.server.routing.*
import java.sql.Connection
import java.util.concurrent.atomic.AtomicInteger

/** 全局计数器：用于观察不同接口的调用次数 */
val callCounts = java.util.concurrent.ConcurrentHashMap<String, AtomicInteger>()

fun bump(name: String) {
    callCounts.computeIfAbsent(name) { AtomicInteger(0) }.incrementAndGet()
}

fun Application.bottleneckModule(ds: () -> Connection) {

    routing {

        // ── 埋雷 ①：CPU 密集（热路径上重复编译正则）────────────
        // 症状：CPU 高、P99 高、cpu 火焰图顶部是正则编译
        get("/slow-cpu") {
            bump("slow-cpu")
            val text = "x".repeat(20_000)
            var hits = 0
            repeat(200) {
                val re = Regex("""\d{3}-\d{4}""")     // ❌ 每次编译
                hits += re.findAll(text).count()
            }
            call.respondText("hits=$hits")
        }

        // ── 埋雷 ②：阻塞 IO（在 Default 上执行 JDBC）───────────
        // 症状：CPU 低、所有接口一起慢、cpu 火焰图很空、wall 图全是阻塞栈
        get("/slow-io") {
            bump("slow-io")
            val n = ds().use { c ->
                c.createStatement().use { s ->
                    s.executeQuery("select pg_sleep(0.02), 1").use { rs ->
                        rs.next(); rs.getInt(2)
                    }
                }
            }
            call.respondText("n=$n")
        }

        // ── 埋雷 ③：N+1（循环里单条查询）──────────────────────
        // 症状：DB calls 异常高、火焰图看不出什么
        get("/slow-db") {
            bump("slow-db")
            val ids = (1L..100L).map { (it * 7919) % 100_000 + 1 }
            val urls = ids.map { id ->
                ds().connection.use { c ->                // ❌ 每次借还连接 + 单条查询
                    c.prepareStatement("select url from orders where id = ?").use { ps ->
                        ps.setLong(1, id)
                        ps.executeQuery().use { rs -> if (rs.next()) rs.getString(1) else null }
                    }
                }
            }
            call.respondText(urls.filterNotNull().joinToString("\n"))
        }

        // ── 对照接口：正常 ────────────────────────────────────
        get("/normal") {
            bump("normal")
            call.respondText("ok")
        }

        // ── 调用次数指标（用于验证"只压了一个接口"）────────────
        get("/debug/callcounts") {
            call.respondText(
                callCounts.entries.joinToString("\n") { "${it.key}=${it.value.get()}" }
            )
        }
    }
}

关键：三个埋雷分别独立（不同路由），这样分批施压时症状才是干净的。

二、单接口压测脚本

// loadtest/single-endpoint.js
import http from 'k6/http';
import { check } from 'k6';

const ENDPOINT = __ENV.ENDPOINT || 'normal';

export const options = {
  scenarios: {
    single: {
      executor: 'constant-arrival-rate',
      rate: Number(__ENV.RATE || 100),
      timeUnit: '1s',
      duration: __ENV.DURATION || '3m',
      preAllocatedVUs: 200,
      maxVUs: 2000,
    },
  },
  discardResponseBodies: true,
  summaryTrendStats: ['avg', 'med', 'p(95)', 'p(99)', 'max'],
};

export default function () {
  const res = http.get(`${__ENV.BASE_URL}/${ENDPOINT}`, {
    tags: { name: ENDPOINT },
  });
  check(res, { ok: (r) => r.status === 200 });
}

三、分批取证（Lab 6 的核心）

#!/usr/bin/env bash
# tools/lab6-capture.sh <EXP_ID>
#
# 对三个埋雷接口【分别】施压并采证。
# 关键：一次只压一个接口（单变量原则）。
set -uo pipefail

EXP_ID="${1:?usage: lab6-capture.sh <exp_id>}"
BASE_DIR="docs/experiments/${EXP_ID}/results"
mkdir -p "$BASE_DIR"

PID=$(jcmd 2>/dev/null | grep app.jar | awk '{print $1}')
[ -z "$PID" ] && { echo "❌ 找不到应用进程"; exit 1; }

echo "═══════════════════════════════════════════════════════════════"
echo "Lab 6 分批取证（PID=$PID）"
echo "═══════════════════════════════════════════════════════════════"
echo

for EP in slow-cpu slow-io slow-db normal; do
  DIR="$BASE_DIR/$EP"
  mkdir -p "$DIR"

  echo "───────── 接口：/$EP ─────────"

  # 记录施压前的调用次数（用于验证"只压了这一个"）
  curl -s localhost:8080/debug/callcounts > "$DIR/callcounts-before.txt" 2>/dev/null || true

  # ① 后台施压
  BASE_URL=http://127.0.0.1:8080 ENDPOINT="$EP" RATE=100 DURATION=3m \
  k6 run --summary-export="$DIR/k6-summary.json" \
         loadtest/single-endpoint.js > "$DIR/k6-stdout.txt" 2>&1 &
  K6_PID=$!

  echo "  等待 40 秒进入稳态..."
  sleep 40

  # ② 四类火焰图（每类 20 秒，注意：串行采集会让总时长变长）
  echo "  采集火焰图..."
  for E in cpu wall alloc lock; do
    asprof -d 20 -e "$E" -f "$DIR/$E.html" "$PID" > /dev/null 2>&1
    asprof -d 20 -e "$E" -o collapsed -f "$DIR/$E.collapsed" "$PID" > /dev/null 2>&1
  done

  # ③ 线程快照（连续 3 份）
  echo "  采集线程快照..."
  mkdir -p "$DIR/threads"
  for i in 1 2 3; do
    jcmd "$PID" Thread.print > "$DIR/threads/threads-$i.txt" 2>/dev/null
    sleep 2
  done
  python3 tools/analyze-threads.py "$DIR/threads"/threads-*.txt > "$DIR/threads/analysis.txt" 2>&1

  # ④ 指标快照
  curl -s localhost:8080/metrics > "$DIR/metrics.txt" 2>/dev/null || true

  # ⑤ CPU 利用率（这个接口的特征）
  top -b -n 1 -p "$PID" 2>/dev/null | tail -1 | awk '{print "cpu_pct=" $9}' > "$DIR/cpu.txt" || true

  # ⑥ 数据库侧（如果有 PG_CONN）
  if [ -n "${PG_CONN:-}" ]; then
    psql "$PG_CONN" -tAc "
      SELECT calls || ',' || round(mean_exec_time::numeric,2) || ',' ||
             round(total_exec_time::numeric,0) || ',' ||
             left(regexp_replace(query,'\s+',' ','g'), 50)
      FROM pg_stat_statements ORDER BY calls DESC LIMIT 10;" > "$DIR/pg-stats.txt" 2>/dev/null || true
  fi

  # 记录施压后的调用次数
  curl -s localhost:8080/debug/callcounts > "$DIR/callcounts-after.txt" 2>/dev/null || true

  wait $K6_PID 2>/dev/null || true

  # 打印摘要
  echo "  ├─ CPU: $(cat "$DIR/cpu.txt" 2>/dev/null || echo 'n/a')"
  if [ -f "$DIR/k6-summary.json" ]; then
    python3 - "$DIR/k6-summary.json" <<'PY'
import json, sys
try:
    m = json.load(open(sys.argv[1]))["metrics"]
    d = m["http_req_duration"]
    print(f"  ├─ P50={d['med']:.1f}ms  P99={d['p(99)']:.1f}ms  错误率={m['http_req_failed']['rate']:.2%}")
except Exception as e:
    print(f"  ├─ （解析失败: {e}）")
PY
  fi
  echo "  └─ 证据 → $DIR"
  echo

  sleep 10   # 接口之间留间隔，避免互相影响
done

echo "═══════════════════════════════════════════════════════════════"
echo "✅ 取证完成"
echo
echo "下一步：生成证据矩阵"
echo "  python3 tools/lab6-evidence-matrix.py $EXP_ID"
echo "═══════════════════════════════════════════════════════════════"

四、证据矩阵自动生成

# tools/lab6-evidence-matrix.py <EXP_ID>
"""
扫描各接口的证据，生成"证据矩阵"。

矩阵的每一列是一个接口，每一行是一种工具 —— 它会清楚地显示：
【哪个瓶颈只在哪种工具上可见】
"""
import json
import pathlib
import re
import sys


def read(path):
    try:
        return pathlib.Path(path).read_text(encoding="utf-8", errors="ignore")
    except Exception:
        return ""


def top_frames(collapsed_path, n=3):
    """从 collapsed 文件里找占比最高的叶子帧"""
    text = read(collapsed_path)
    if not text.strip():
        return []
    from collections import Counter
    self_time = Counter()
    for line in text.splitlines():
        if not line.strip():
            continue
        try:
            *frames, count = line.rsplit(" ", 1)
            self_time[frames[0].split(";")[-1]] += int(count)
        except (ValueError, IndexError):
            continue
    total = sum(self_time.values())
    if total == 0:
        return []
    return [(f.split(".")[-1][:28], c / total * 100) for f, c in self_time.most_common(n)]


def cpu_pct(path):
    text = read(path)
    m = re.search(r"cpu_pct=([\d.]+)", text)
    return float(m.group(1)) if m else None


def thread_summary(path):
    text = read(path)
    states = re.findall(r"^\s+([A-Z_]+)\s+\d+\s+[\d.]+%", text, re.MULTILINE)
    return "/".join(states[:2]) if states else "?"


def db_calls(path):
    text = read(path)
    if not text.strip():
        return None
    max_calls = 0
    for line in text.splitlines():
        parts = line.split(",")
        if parts and parts[0].strip().isdigit():
            max_calls = max(max_calls, int(parts[0]))
    return max_calls


def k6_p99(path):
    try:
        m = json.loads(read(path))["metrics"]
        return m["http_req_duration"]["p(99)"]
    except Exception:
        return None


def main(exp_id):
    base = pathlib.Path("docs/experiments") / exp_id / "results"
    endpoints = ["slow-cpu", "slow-io", "slow-db", "normal"]

    print("═" * 100)
    print(f"Lab 6 证据矩阵：{exp_id}")
    print("═" * 100)
    print()

    rows = []
    for ep in endpoints:
        d = base / ep
        if not d.exists():
            continue
        rows.append({
            "ep": ep,
            "cpu_pct": cpu_pct(d / "cpu.txt"),
            "p99": k6_p99(d / "k6-summary.json"),
            "cpu_top": top_frames(d / "cpu.collapsed"),
            "wall_top": top_frames(d / "wall.collapsed"),
            "alloc_top": top_frames(d / "alloc.collapsed"),
            "lock_top": top_frames(d / "lock.collapsed"),
            "threads": thread_summary(d / "threads" / "analysis.txt"),
            "db_calls": db_calls(d / "pg-stats.txt"),
        })

    if not rows:
        print("❌ 没有找到证据（先运行 tools/lab6-capture.sh）")
        return

    # 基础指标表
    print("【基础指标】")
    print(f"{'接口':<12}{'CPU%':>8}{'P99(ms)':>10}{'线程状态':>16}{'DB calls':>12}")
    print("-" * 100)
    for r in rows:
        print(f"{'/' + r['ep']:<12}"
              f"{(f'{r[chr(99)+chr(112)+chr(117)+chr(95)+chr(112)+chr(99)+chr(116)]:.0f}' if r['cpu_pct'] is not None else 'n/a'):>8}"
              f"{(f'{r[chr(112)+chr(57)+chr(57)]:.1f}' if r['p99'] else 'n/a'):>10}"
              f"{r['threads']:>16}"
              f"{(str(r['db_calls']) if r['db_calls'] else 'n/a'):>12}")
    print()

    # 火焰图 Top 帧
    for key, label in [("cpu_top", "cpu 火焰图"), ("wall_top", "wall 火焰图"),
                       ("alloc_top", "alloc 火焰图"), ("lock_top", "lock 火焰图")]:
        print(f"【{label}】最宽的自耗时帧")
        for r in rows:
            tops = r[key]
            if tops:
                desc = ", ".join(f"{f}({p:.0f}%)" for f, p in tops[:2])
            else:
                desc = "（空 / 无数据）"
            print(f"  /{r['ep']:<10} {desc}")
        print()

    # 结论提示
    print("═" * 100)
    print("【判读：哪个瓶颈在哪种工具上可见】")
    print()
    print("  预期（做完后对照）：")
    print("    /slow-cpu → cpu 火焰图有明显热点（正则编译）；CPU 高")
    print("    /slow-io  → cpu 图【很空】；wall 图有大量阻塞栈；CPU 低")
    print("    /slow-db  → 火焰图看不出什么；但 DB calls 异常高 ⭐")
    print()
    print("  如果你只采 CPU 火焰图，你只能发现第一个瓶颈。")
    print("  —— 这就是本章的核心教学点。")
    print("═" * 100)


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("usage: lab6-evidence-matrix.py <EXP_ID>")
        sys.exit(1)
    main(sys.argv[1])

注意：上面的 cpu_pct 那行用了 chr() 拼接来避免字典键里的特殊字符问题，实际使用时可简化为 r['cpu_pct']——这里保留是为了演示当脚本需要处理不确定的键名时的一种防御写法。

五、Lab 6 验收

#!/usr/bin/env bash
# tools/verify-lab6.sh <EXP_ID>
set -uo pipefail

EXP_ID="${1:?usage: verify-lab6.sh <exp_id>}"
BASE="docs/experiments/${EXP_ID}"
PASS=0; FAIL=0

ok()  { echo "  ✅ $1"; PASS=$((PASS+1)); }
bad() { echo "  ❌ $1"; FAIL=$((FAIL+1)); }

echo "═══ Lab 6 验收：$EXP_ID ═══"
echo

echo "① 三个埋雷接口的证据完整性"
for EP in slow-cpu slow-io slow-db; do
  D="$BASE/results/$EP"
  if [ ! -d "$D" ]; then
    bad "$EP：目录不存在"
    continue
  fi
  MISSING=""
  for F in k6-summary.json cpu.collapsed wall.collapsed metrics.txt cpu.txt; do
    [ -f "$D/$F" ] || MISSING="$MISSING $F"
  done
  [ -f "$D/threads/analysis.txt" ] || MISSING="$MISSING threads/analysis.txt"

  if [ -z "$MISSING" ]; then
    ok "$EP：证据齐全"
  else
    bad "$EP：缺少$MISSING"
  fi
done
echo

echo "② 关键差异检查（三个接口的 CPU 应该有明显区别）"
python3 - "$BASE" <<'PY'
import pathlib, re, sys
base = pathlib.Path(sys.argv[1])
cpus = {}
for ep in ("slow-cpu", "slow-io", "slow-db"):
    p = base / "results" / ep / "cpu.txt"
    if p.exists():
        m = re.search(r"cpu_pct=([\d.]+)", p.read_text(errors="ignore"))
        if m:
            cpus[ep] = float(m.group(1))
if len(cpus) >= 2:
    for ep, v in cpus.items():
        print(f"     /{ep}: CPU {v:.0f}%")
    high = max(cpus.values()); low = min(cpus.values())
    if high - low > 15:
        print(f"     ✅ CPU 差异明显（{high-low:.0f} 个百分点）—— 说明埋雷设计有效")
    else:
        print(f"     ⚠️  CPU 差异不明显（{high-low:.0f} 个百分点）")
        print(f"        可能原因：压测强度不够 / 埋雷不够重 / 接口之间有互相影响")
else:
    print("     ⚠️  数据不足，无法比较")
PY
echo

echo "③ 三份根因结论"
if [ -f "$BASE/conclusions.md" ]; then
  CNT=$(grep -c "^## " "$BASE/conclusions.md" || echo 0)
  [ "$CNT" -ge 3 ] && ok "conclusions.md 含 $CNT 个章节" || bad "conclusions.md 章节不足（$CNT）"
  # 检查必需要素
  for KW in "证据" "贡献占比" "被排除"; do
    grep -q "$KW" "$BASE/conclusions.md" && ok "含「$KW」" || bad "缺「$KW」"
  done
else
  bad "缺少 conclusions.md（本 Lab 的核心产出）"
fi
echo

echo "④ 证据矩阵"
[ -f "$BASE/evidence-matrix.txt" ] && ok "证据矩阵已生成" || bad "未生成证据矩阵（跑 lab6-evidence-matrix.py）"
echo

echo "⑤ 实验档案"
[ -f "$BASE/README.md" ] && ok "README.md 存在" || bad "缺 README.md"
echo

echo "═══════════════════════════════"
echo "通过 $PASS 项，失败 $FAIL 项"
[ "$FAIL" -eq 0 ] && echo "✅ Lab 6 完成" || echo "❌ 还有 $FAIL 项待补"

六、动手改造

改动 观察什么
加大 /slow-cpu 的循环次数 CPU 利用率上升，但 wall 图的比例不变——理解"纯 CPU 瓶颈"的特征
把 /slow-io 的 JDBC 切到 Dispatchers.IO 重跑取证，wall 图里的阻塞栈消失，卡顿恢复
给 /slow-db 改成批量查询 DB calls 从 100/请求降到 1/请求，P99 大幅下降
同时压三个接口 证据矩阵会变得"脏"——这就是为什么必须分批施压
只采 cpu 火焰图，跳过 wall 你只能发现 /slow-cpu——亲手验证"漏掉 2/3 瓶颈"

七、这段代码的局限

  • lab6-capture.sh 耗时较长（4 个接口 × 约 2.5 分钟 ≈ 10 分钟 + 火焰图采集时间）：因为四类火焰图是串行采集的（每类 20 秒，共 80 秒/接口）。
  • 火焰图串行采集会让"施压窗口"变长：如果服务状态在采集期间变化，各张图的条件可能不一致。改进方法：缩短每类的时长（-d 10）。
  • lab6-evidence-matrix.py 的解析依赖文件格式：collapsed 格式和 k6 summary 的字段名可能因版本而异。
  • /slow-db 需要预先有 orders 表与数据：如果没有，slow-db 接口会报错而不是变慢。
  • 本 Lab 的结论是"教学性质"的：真实排查中，三个瓶颈很少这么"干净"地分开——所以分批施压（单变量）这个习惯比脚本本身更重要。