6.6 配套代码：数据与算法类瓶颈诊断
对应小节：6.6 瓶颈模式库（二）：数据与算法类 八种模式各有对应的诊断查询——这份文件把它们集中起来，一条命令全查一遍。
一、八合一诊断脚本
#!/usr/bin/env bash
# tools/diagnose-data-bottlenecks.sh <PG_CONN> [METRICS_URL] [PID]
set -uo pipefail
PG="${1:-${PG_CONN:-}}"
URL="${2:-http://127.0.0.1:8080/metrics}"
PID="${3:-$(jcmd 2>/dev/null | grep app.jar | awk '{print $1}')}"
echo "═══════════════════════════════════════════════════════════════"
echo "数据与算法类瓶颈诊断"
echo "═══════════════════════════════════════════════════════════════"
echo
# ═══ 模式一：N+1 查询 ═══════════════════════════════════════
echo "【模式一：N+1 查询】"
echo " 判据：某简单查询的 calls 远大于接口 QPS"
if [ -n "$PG" ]; then
psql "$PG" -tAc "
SELECT ' calls=' || calls || ' | mean=' || round(mean_exec_time::numeric,2) || 'ms'
|| ' | total=' || round(total_exec_time::numeric,0) || 'ms | '
|| left(regexp_replace(query,'\s+',' ','g'), 55)
FROM pg_stat_statements
WHERE query NOT LIKE '%pg_stat_statements%'
ORDER BY calls DESC LIMIT 8;" 2>/dev/null || echo " （无法查询）"
echo
echo " ⭐ 关键：如果某个「单次很快」的查询 calls 极高 → N+1"
echo " 修法：改成批量查询（where id = any(?)）"
else
echo " （未提供 PG_CONN，跳过）"
fi
echo
# ═══ 模式二：缺索引 / 全表扫描 ══════════════════════════════
echo "【模式二：缺索引 / 全表扫描】"
if [ -n "$PG" ]; then
psql "$PG" -tAc "
SELECT ' ' || relname || ': seq_scan=' || seq_scan || ' idx_scan=' || COALESCE(idx_scan,0)
|| ' rows=' || n_live_tup
FROM pg_stat_user_tables
WHERE seq_scan > 1000 AND n_live_tup > 10000
ORDER BY seq_scan DESC LIMIT 5;" 2>/dev/null || echo " （无法查询）"
echo
echo " ⭐ 高 seq_scan + 大表 → 可能有查询没走索引"
echo " 验证：EXPLAIN (ANALYZE, BUFFERS) <你的查询>"
echo " 危险信号：Seq Scan / Rows Removed by Filter 很大 / 行数估计偏差大"
else
echo " （未提供 PG_CONN，跳过）"
fi
echo
# ═══ 模式三：锁竞争 ═════════════════════════════════════════
echo "【模式三：锁竞争】"
if [ -n "$PG" ]; then
echo " 数据库侧的阻塞链："
psql "$PG" -tAc "
SELECT ' blocked=' || blocked.pid || ' by=' || blocking.pid
|| ' | ' || left(regexp_replace(blocked.query,'\s+',' ','g'), 40)
FROM pg_stat_activity blocked
JOIN pg_stat_activity blocking ON blocking.pid = ANY(pg_blocking_pids(blocked.pid))
LIMIT 5;" 2>/dev/null || echo " （无法查询）"
fi
if [ -n "$PID" ]; then
TMP=$(mktemp)
jcmd "$PID" Thread.print > "$TMP" 2>/dev/null
BLOCKED=$(grep -c "java.lang.Thread.State: BLOCKED" "$TMP" || echo 0)
echo " 应用侧的 BLOCKED 线程数: $BLOCKED"
[ "$BLOCKED" -gt 5 ] && echo " ❌ 大量 BLOCKED → 应用内锁竞争" || echo " ✅ 无明显锁竞争"
rm -f "$TMP"
fi
echo
# ═══ 模式四：GC 停顿 ════════════════════════════════════════
echo "【模式四：GC 停顿】"
M=$(curl -s --max-time 5 "$URL" 2>/dev/null || echo "")
if [ -n "$M" ]; then
GC_P99=$(echo "$M" | awk '/jvm_gc_pause_seconds/{print $2}' | sort -rn | head -1)
ALLOC=$(echo "$M" | awk '/jvm_gc_memory_allocated_bytes_total/{printf "%.1f", $2/1048576; exit}')
echo " GC 暂停相关指标: ${GC_P99:-n/a}"
echo " 累计分配: ${ALLOC:-n/a} MB"
echo
echo " ⭐ 唯一确定的验证：把 gc.log 的停顿时间戳与 P99 尖刺时间戳【对齐】"
echo " 对齐 → GC 是根因；不对齐 → 排除 GC（能省大量时间）"
else
echo " （无法获取指标）"
fi
echo
# ═══ 模式五：缓存击穿/雪崩 ══════════════════════════════════
echo "【模式五：缓存击穿 / 雪崩】"
if command -v redis-cli > /dev/null 2>&1; then
redis-cli INFO stats 2>/dev/null | grep -E "keyspace_(hits|misses)" | sed 's/^/ /' || true
echo " ⭐ 判据：DB QPS 出现周期性脉冲，且与缓存过期时间吻合"
echo " 修法：TTL 加抖动（防雪崩）+ 单飞（防击穿）+ 缓存空值（防穿透）"
else
echo " （redis-cli 不可用）"
fi
echo
# ═══ 模式六：序列化开销 ═════════════════════════════════════
echo "【模式六：序列化开销】"
echo " 验证：采 CPU 火焰图，看顶部是否有很宽的序列化栈"
echo " asprof -d 60 -e cpu -f cpu.html ${PID:-<pid>}"
echo " 常见栈帧：Jackson.writeValue / StringBuilder / String.format / Integer.valueOf"
echo " ⭐ 优先考虑「减少传输的数据量」，而不是「换序列化器」"
echo
# ═══ 模式七：日志同步 IO ════════════════════════════════════
echo "【模式七：日志同步 IO】"
if [ -n "$PID" ]; then
TMP=$(mktemp)
jcmd "$PID" Thread.print > "$TMP" 2>/dev/null
FILEIO=$(grep -c -E "FileOutputStream.write|FileDispatcherImpl.write" "$TMP" || echo 0)
echo " 停在文件写入的线程数: $FILEIO"
[ "$FILEIO" -gt 0 ] && echo " ⚠️ 有线程在写文件 → 检查是否同步日志" || echo " ✅ 无"
rm -f "$TMP"
fi
echo " ⭐ 最确定的验证：JFR 的 jdk.FileWrite 事件"
echo " jfr print --events jdk.FileWrite recording.jfr | grep -c app.log"
echo " 对照实验：把日志级别降到 ERROR，看 P99 是否回落"
echo
# ═══ 模式八：无界队列 OOM ═══════════════════════════════════
echo "【模式八：无界队列 OOM】"
if [ -n "$M" ]; then
Q=$(echo "$M" | awk '/executor_queue_depth/{printf "%.0f", $2; exit}')
HEAP=$(echo "$M" | awk '/jvm_memory_used_bytes.*heap/{printf "%.0f", $2/1048576; exit}')
echo " 队列深度: ${Q:-n/a}"
echo " 堆使用 : ${HEAP:-n/a} MB"
echo " ⭐ 判据：堆的「GC 后基线」单调上升（不是瞬时值）"
echo " 用第 3 章 3.5 的 analyze_soak.py 判断趋势"
fi
echo
echo "═══════════════════════════════════════════════════════════════"
echo "八种模式的「一个指标区分」速查："
echo " N+1 vs 缺索引 ：看 calls（高频）还是 mean（单次慢）"
echo " 锁竞争 vs CPU 饱和 ：sys 占比高 + BLOCKED 多 → 锁"
echo " GC vs 其他周期性 ：时间戳对齐（唯一确定的方法）"
echo " 缓存击穿 vs 雪崩 ：单 key 脉冲 vs 大批 key 同时失效"
echo " 日志 IO vs 网络 IO ：JFR 的 FileWrite vs SocketWrite"
echo "═══════════════════════════════════════════════════════════════"
二、N+1 的专项检测（代码侧）
// src/main/kotlin/metrics/QueryCounter.kt
package metrics
import io.micrometer.core.instrument.Counter
import io.micrometer.core.instrument.MeterRegistry
import io.micrometer.core.instrument.Timer
import kotlinx.coroutines.ThreadContextElement
import kotlin.coroutines.AbstractCoroutineContextElement
import kotlin.coroutines.CoroutineContext
/**
* 统计"每个请求执行了多少次数据库查询"。
* 这是发现 N+1 最直接的手段——比看 pg_stat_statements 更早、更准。
*
* 用法：
* withQueryCounting { handleRequest() } // 返回 (结果, 查询次数)
* 或者把计数作为指标暴露出去
*/
class QueryCounter(registry: MeterRegistry) {
private val queriesPerRequest = Timer.builder("app.db.queries.per.request")
.description("每个请求的数据库查询次数（N+1 检测）")
.publishPercentileHistogram(false)
.register(registry)
// 用一个可变的 holder 在协程/线程上下文中传递
class Holder : AbstractCoroutineContextElement(Holder) {
companion object Key : CoroutineContext.Key<Holder>
var count: Int = 0
}
/** 在数据库调用处调用这个方法 */
fun countQuery() {
// 在协程场景下，通过上下文传递；在线程场景下用 ThreadLocal
}
/** 包装一次请求处理，记录查询次数 */
suspend fun <T> countQueries(block: suspend () -> T): Pair<T, Int> {
val holder = Holder()
val result = kotlinx.coroutines.withContext(holder) { block() }
queriesPerRequest.record(holder.count.toDouble())
return result to holder.count
}
}
/**
* 更简单的实现：用 ThreadLocal（如果不用协程）
*/
object SimpleQueryCounter {
private val counter = ThreadLocal.withInitial { intArrayOf(0) }
fun reset() { counter.get()[0] = 0 }
fun increment() { counter.get()[0]++ }
fun get(): Int = counter.get()[0]
}
在 Repository 层埋点：
class OrderRepository(private val ds: DataSource) {
fun findById(id: Long): Order? {
SimpleQueryCounter.increment() // ← 每执行一次查询就 +1
return ds.connection.use { c -> /* ... */ }
}
fun findByIds(ids: List<Long>): Map<Long, Order> {
SimpleQueryCounter.increment() // ← 批量查询只 +1
return ds.connection.use { c -> /* ... */ }
}
}
在接口层验证：
get("/orders") {
SimpleQueryCounter.reset()
val orders = orderService.list() // 内部可能触发 N+1
val queryCount = SimpleQueryCounter.get()
if (queryCount > 10) {
log.warn("可能的 N+1：本次请求执行了 {} 次查询", queryCount)
}
call.respond(orders)
}
最佳实践：把这个计数做成指标（app_db_queries_per_request），设一个 P99 告警——一旦某次改动引入了 N+1，指标立刻能看出来。
三、GC 时间戳对齐（脚本化）
# tools/align-gc-with-latency.py <gc.log> <latency.csv>
"""
把 GC 停顿时间戳与延迟尖刺时间戳对齐。
这是判断"GC 是不是根因"的【唯一确定性方法】。
输入：
gc.log PostgreSQL/JVM 的 GC 日志（含时间戳与停顿时长）
latency.csv 时间戳,延迟(ms) 两列
"""
import csv
import re
import sys
from datetime import datetime
def parse_gc_pauses(path):
"""从 GC 日志提取 (时间戳秒, 停顿毫秒)"""
pauses = []
with open(path, errors="ignore") as f:
for line in f:
if "Pause Young" not in line and "Pause Full" not in line:
continue
# 提取 uptime（形如 [12.456s]）或时间戳
m = re.search(r"\[([\d.]+)s\]", line)
if not m:
continue
uptime = float(m.group(1))
# 提取末尾的停顿时间（形如 4.231ms）
pm = re.search(r"([\d.]+)ms\s*$", line.strip())
pause_ms = float(pm.group(1)) if pm else 0.0
pauses.append((uptime, pause_ms))
return pauses
def parse_latency(path):
"""从 CSV 读取 (时间戳秒, 延迟ms)"""
rows = []
with open(path) as f:
reader = csv.reader(f)
for row in reader:
if len(row) < 2:
continue
try:
rows.append((float(row[0]), float(row[1])))
except ValueError:
continue
return rows
def main(gc_path, latency_path, window_s=1.0):
pauses = parse_gc_pauses(gc_path)
latencies = parse_latency(latency_path)
if not pauses or not latencies:
print(f"❌ 数据不足：GC 停顿 {len(pauses)} 条，延迟 {len(latencies)} 条")
print(" 提示：GC 日志需要包含 uptime（-Xlog:gc*:time,uptime）")
return
# 找出延迟尖刺（超过 P95 的点）
vals = sorted(v for _, v in latencies)
p95 = vals[int(len(vals) * 0.95)]
spikes = [(t, v) for t, v in latencies if v > p95]
print("═" * 74)
print("GC 停顿 vs 延迟尖刺 对齐分析")
print("═" * 74)
print()
print(f"GC 停顿数 : {len(pauses)}")
print(f"延迟样本数 : {len(latencies)}")
print(f"P95 阈值 : {p95:.1f} ms")
print(f"延迟尖刺数 : {len(spikes)}")
print()
if not spikes:
print("⚠️ 没有延迟尖刺（所有样本都在 P95 以下）")
return
# 对齐检测
matched = 0
for st, sv in spikes[:20]:
for pt, pv in pauses:
if abs(st - pt) <= window_s:
matched += 1
break
checked = min(20, len(spikes))
ratio = matched / checked * 100 if checked else 0
print(f"对齐检查（窗口 ±{window_s}s）：")
print(f" 检查了 {checked} 个尖刺，其中 {matched} 个与 GC 停顿时间吻合（{ratio:.0f}%）")
print()
if ratio > 70:
print("✅ 高度对齐 → GC 停顿是延迟尖刺的根因")
print()
print("下一步：")
print(" ① 看停顿幅度（{:.1f}ms 级别的停顿能不能解释尖刺高度）".format(
max(p for _, p in pauses)))
print(" ② 采 alloc 火焰图找分配热点")
print(" ③ 检查堆大小与 GC 选型")
elif ratio > 30:
print("⚠️ 部分对齐 → GC 可能是原因之一，但不是全部")
print(" 建议：把 GC 停顿的贡献占比单独算出来")
else:
print("❌ 不对齐 → 排除 GC")
print()
print("这是一个重要的排除结论 —— 能省下大量时间。")
print("下一步转向：连接池、慢查询、锁竞争、下游调用、容器节流")
print("═" * 74)
if __name__ == "__main__":
if len(sys.argv) < 3:
print("usage: align-gc-with-latency.py <gc.log> <latency.csv> [window_s]")
sys.exit(1)
main(sys.argv[1], sys.argv[2],
float(sys.argv[3]) if len(sys.argv) > 3 else 1.0)
四、动手改造
| 改动 | 观察什么 |
|---|---|
在 Lab 6 的三个接口上分别跑 diagnose-data-bottlenecks.sh |
只有 /slow-db 会显示 calls 异常 |
| 故意去掉一个索引并压测 | 模式二的 seq_scan 会飙升 |
用 align-gc-with-latency.py 跑一份真实数据 |
体会「对齐 → 确认」「不对齐 → 排除」两种结论 |
把 QueryCounter 接进你的 Repository 层 |
得到一个能提前发现 N+1 的指标 |
| 故意让某个查询 sleep 3 秒并观察阻塞链查询 | 验证模式三的检测能力 |
五、这段代码的局限
diagnose-data-bottlenecks.sh需要pg_stat_statements与 psql 权限：没有这些会跳过对应检查。align-gc-with-latency.py需要 GC 日志包含 uptime：用-Xlog:gc*:time,uptime才有；如果只有绝对时间戳，需要改解析逻辑。- 对齐的时间窗口（默认 ±1 秒）需要调整：如果延迟采样粒度是 10 秒，窗口要放大；如果是 1 秒，窗口可以缩小。
QueryCounter的协程版本还不完整：Holder需要在数据库调用处能拿到（用coroutineContext[Holder]），实现比 ThreadLocal 复杂。简单场景用 ThreadLocal 版本更实用。- 脚本不能替代判断：它给出的是「哪些模式可能」，最终还要人来看数据、下结论。