文档目录

1.6 配套代码：阶梯加压、拐点检测与容量推算

对应小节：1.6 基线与容量曲线 三件事：① 压出曲线 ② 找到拐点 ③ 推出安全容量与副本数。

一、阶梯加压脚本（k6）

// loadtest/capacity-staircase.js
import http from 'k6/http';
import { check } from 'k6';

// 阶梯：每一级"爬升 30s + 稳态 120s"
// 稳态窗口才是用来采数据的，爬升期不计入
const LEVELS = [100, 200, 400, 700, 1000, 1500, 2000];

const stages = [];
for (const target of LEVELS) {
  stages.push({ target, duration: '30s' });    // 爬升
  stages.push({ target, duration: '2m' });     // 稳态（采数据）
}

export const options = {
  scenarios: {
    staircase: {
      executor: 'ramping-arrival-rate',
      startRate: 50,
      timeUnit: '1s',
      preAllocatedVUs: 500,
      maxVUs: 5000,                             // 必须给够，否则实际到达率低于设定值
      stages,
    },
  },
  // 这里故意不设 thresholds：容量测试的目的是"找到拐点"，不是"判定通过/失败"
  discardResponseBodies: true,
};

export default function () {
  // 用幂律分布取样，制造真实的热点
  const u = Math.random();
  const id = Math.max(1, Math.floor(1_000_000 * Math.pow(u, 3)) + 1);
  const res = http.get(`${__ENV.BASE_URL}/orders/${id}?stage=${__ENV.STAGE ?? 'unknown'}`);
  check(res, { ok: (r) => r.status === 200 || r.status === 404 });
}

# 运行并把原始数据落盘
k6 run --out json=docs/experiments/E02-capacity/results/k6-raw.json \
       --summary-export=docs/experiments/E02-capacity/results/k6-summary.json \
       loadtest/capacity-staircase.js

关键检查：跑完后先看实际 QPS 是否等于设定值。如果 k6 报告的到达率低于 LEVELS，说明 preAllocatedVUs 不够——此时曲线不可用（第 0 章 0.6 节的协调遗漏）。

二、拐点检测

先说清一件事：拐点检测是启发式的，最终必须人工看一眼曲线。 下面这个脚本给你一个候选点，不是标准答案。

# tools/find_knee.py
"""
从 k6 的分段结果里找拐点。
思路：把每一级的 P99 与 QPS 画成曲线后，"拐点"是延迟增长率开始显著超过负载增长率的那个位置。

判据：growth_ratio = (p99[i]/p99[0]) / (qps[i]/qps[0])
      growth_ratio 明显 > 1 说明"延迟涨得比负载快" → 排队开始显现
"""
import json, sys, pathlib

def load_stages(path):
    """从 k6 原始 JSON 里按 stage 标签聚合（简化示例，实际按时间窗切片）"""
    rows = []
    with open(path) as f:
        for line in f:
            try:
                d = json.loads(line)
            except json.JSONDecodeError:
                continue
            if d.get("type") == "Point" and d.get("metric") == "http_req_duration":
                rows.append((d["data"]["time"], d["data"]["value"]))
    rows.sort()
    return rows

def summarize(rows, windows=7):
    """把时间轴等分成若干窗口，每个窗口算 P50/P99（简化：用最大近似 P99 上界）"""
    if not rows:
        return []
    n = len(rows) // windows
    out = []
    for i in range(windows):
        chunk = [v for _, v in rows[i * n:(i + 1) * n]]
        if not chunk:
            continue
        chunk.sort()
        out.append({
            "p50": chunk[len(chunk) // 2],
            "p99": chunk[min(int(len(chunk) * 0.99), len(chunk) - 1)],
        })
    return out

def find_knee(stats, qps_levels):
    """
    返回 (拐点索引, 每个点的 growth_ratio)
    growth_ratio 明显大于 1 → 延迟增长快于负载增长 → 排队开始
    """
    if not stats:
        return -1, []
    base_p99 = stats[0]["p99"] or 1e-9
    base_qps = qps_levels[0]
    ratios = []
    for i, s in enumerate(stats):
        qps = qps_levels[min(i, len(qps_levels) - 1)]
        latency_growth = s["p99"] / base_p99
        load_growth = qps / base_qps
        ratios.append(latency_growth / load_growth)

    # 拐点 = growth_ratio 首次明显超过基线（> 1.5）的位置
    for i, r in enumerate(ratios):
        if r > 1.5:
            return i, ratios
    return len(ratios) - 1, ratios

if __name__ == "__main__":
    path = sys.argv[1]
    levels = [int(x) for x in sys.argv[2].split(",")]
    stats = summarize(load_stages(path), windows=len(levels))
    knee, ratios = find_knee(stats, levels)

    print(f"{'窗口':<6}{'P50(ms)':>10}{'P99(ms)':>10}{'延迟/负载增长率':>18}  判断")
    for i, s in enumerate(stats):
        qps = levels[min(i, len(levels) - 1)]
        verdict = ""
        if i == knee:
            verdict = "← 拐点候选"
        elif ratios[i] > 1.5:
            verdict = "已进入排队区"
        print(f"{qps:<6}{s['p50']:>10.1f}{s['p99']:>10.1f}{ratios[i]:>18.2f}  {verdict}")

    print()
    print("⚠️ 这只是候选点。请务必人工看一眼曲线，并对照饱和度指标确认")
    print("   （连接池 pending / 线程池队列 / CPU 节流 是哪一类先动）。")

输出形态：

窗口    P50(ms)   P99(ms)   延迟/负载增长率  判断
100        12.0      45.0              1.00
200        12.5      48.0              1.07
400        13.1      52.0              1.16
700        14.0      61.0              1.36
1000       18.2      95.0              2.11  ← 拐点候选
1500       41.0     310.0              9.17  已进入排队区
2000      120.0    1400.0             62.5   已进入排队区

注意 growth_ratio 从 1.36 跳到 2.11 再跳到 9.17 —— 这个非线性跳跃就是拐点的特征。

三、从拐点推算容量与副本数

// src/main/kotlin/capacity/CapacityPlan.kt
package capacity

import kotlin.math.ceil

data class CapacityInput(
    val kneeQps: Int,                 // 压测找到的拐点（单实例）
    val headroomRatio: Double,        // 建议 0.3 ~ 0.5
    val peakQps: Int,                 // 业务预估峰值
    val redundancy: Int = 1,          // N+1 冗余：坏一台不影响服务
    val instanceSpec: String,
)

data class CapacityPlan(
    val safeCapacityPerInstance: Int,
    val requiredReplicas: Int,
    val utilizationAtPeak: Double,
    val warnings: List<String>,
)

fun plan(input: CapacityInput): CapacityPlan {
    val warnings = mutableListOf<String>()

    if (input.headroomRatio < 0.2) {
        warnings += "headroom %.0f%% 偏小：排队是非线性的，余量不足会在突发时雪崩"
            .format(input.headroomRatio * 100)
    }
    if (input.headroomRatio > 0.6) {
        warnings += "headroom %.0f%% 偏大：成本可能浪费，建议先用压测确认拐点是否测准"
            .format(input.headroomRatio * 100)
    }

    val safe = (input.kneeQps * (1 - input.headroomRatio)).toInt()
    val replicas = ceil(input.peakQps.toDouble() / safe).toInt() + input.redundancy

    warnings += "副本翻倍 ≠ 容量翻倍：共享的数据库/Redis/下游服务会成为新瓶颈，扩容后必须复测"
    warnings += "数据量增长会移动拐点（数据从 100 万到 1000 万，索引效率变化），建议每季度复测"

    return CapacityPlan(
        safeCapacityPerInstance = safe,
        requiredReplicas = replicas,
        utilizationAtPeak = input.peakQps.toDouble() / (safe * replicas),
        warnings = warnings,
    )
}

fun main() {
    println("=== 从拐点推算容量与副本数 ===")
    println()

    val scenarios = listOf(
        CapacityInput(kneeQps = 900, headroomRatio = 0.40, peakQps = 3000, instanceSpec = "4C8G"),
        CapacityInput(kneeQps = 900, headroomRatio = 0.10, peakQps = 3000, instanceSpec = "4C8G"),
        CapacityInput(kneeQps = 900, headroomRatio = 0.40, peakQps = 12000, instanceSpec = "4C8G"),
    )

    scenarios.forEach { input ->
        val p = plan(input)
        println("拐点 %d RPS，headroom %.0f%%，峰值 %d RPS（%s）"
            .format(input.kneeQps, input.headroomRatio * 100, input.peakQps, input.instanceSpec))
        println("  → 单实例安全容量 : %d RPS".format(p.safeCapacityPerInstance))
        println("  → 所需副本数     : %d（含 %d 台冗余）".format(p.requiredReplicas, input.redundancy))
        println("  → 峰值时利用率   : %.0f%%".format(p.utilizationAtPeak * 100))
        p.warnings.forEach { println("  ⚠️  $it") }
        println()
    }
}

输出：

拐点 900 RPS，headroom 40%，峰值 3000 RPS（4C8G）
  → 单实例安全容量 : 540 RPS
  → 所需副本数     : 7（含 1 台冗余）
  → 峰值时利用率   : 79%

拐点 900 RPS，headroom 10%，峰值 3000 RPS（4C8G）
  → 单实例安全容量 : 810 RPS
  → 所需副本数     : 5（含 1 台冗余）
  → 峰值时利用率   : 74%
  ⚠️  headroom 10% 偏小：排队是非线性的，余量不足会在突发时雪崩

注意这两个方案的成本差：headroom 从 40% 降到 10%，机器从 7 台变成 5 台——省了 2 台机器的钱，代价是没有任何余量应对突发。这是一个明确的工程权衡，不是"越省越好"。

四、完整工作流

① 用 staircase 脚本压出曲线
        ↓
② 先验证「实际 QPS == 设定 QPS」（否则数据不可用）
        ↓
③ 用 find_knee.py 找候选拐点
        ↓
④ 人工看曲线 + 对照饱和度指标确认（是哪类资源先动）
        ↓
⑤ 用 plan() 算出安全容量与副本数
        ↓
⑥ 把拐点、headroom、日期、实验编号记进容量表

容量表模板：

实例规格 拐点(实测) headroom 安全容量 峰值 所需副本 数据来源 测量日期
4C8G 900 RPS 40% 540 RPS 3000 7 E02 2025-xx-xx

五、动手改造

改动 观察什么
把 preAllocatedVUs 从 500 改成 20 实际 QPS 会明显低于设定值，曲线完全失真——这就是协调遗漏在实操中的样子
把稳态时长从 2m 改成 20s 曲线会变得很抖，拐点位置每次跑都不一样
把 Find_knee 的阈值从 1.5 改成 1.2 拐点会前移——说明这个阈值是个判断，不是真理，必须人工确认
用均匀分布代替幂律分布再压一次 拐点会明显后移（缓存命中率高、锁竞争少），这就是为什么数据分布必须写进 SLO 前提
给 plan() 加一个「数据量增长系数」 思考：数据量翻 10 倍，拐点大概会怎么变？

六、这段代码的局限

  • 拐点检测是启发式的。真实的曲线形态多种多样（有双拐点、有硬天花板、有缓慢退化），脚本只能给候选，判断必须靠人。
  • 它假设瓶颈是单一资源。如果 CPU 和连接池同时在 700 RPS 附近饱和，曲线会更复杂。
  • 容量不是一次测完就固定的：数据量增长、依赖变化、业务形态变化都会移动拐点。建议每季度或每次大版本发布后复测。
  • 脚本里的 summarize() 是按时间等分窗口的简化实现。严谨做法是让 k6 按 stage 打标签，或直接读 k6-summary.json 的各阶段阈值结果。