7.2 内容产能与排期
上一节把「什么时候发」钉成了版本列车。列车到站日期一旦确定,下一个问题就变成「能不能按期做完」。产业侧那篇 SLG 经济系统运营 2026 清单 把「内容排期」列为运营动作之一,但它停在 checklist 层面;本节把它做成模型——用人天需求与每日产能算完成日,再用蒙特卡洛把「一个数」变成「一个区间」。人天估计本身怎么来(拆解、类比、三点估计)是项目管理常识,这里不展开,重点是把估计喂进可跑的产能模型。
需求、产能与瓶颈
一个版本的内容 backlog 可以拆成若干任务,每个任务给出各职能(策划 design、美术 art、客户端 client、服务端 server、测试 qa)的人天估计。三个量决定了排期:
- 需求:某职能的总人天 = 该职能下所有任务估计之和。
- 每日产能:
headcount × availabilityPct // 100(人天/天)。availabilityPct是有效工时占比,扣掉会议、支持、请假。 - 完成日:
ceil(需求 / 每日产能)。取所有职能完成日的最大值即瓶颈——排期由最慢的职能决定,跟 12 章闭环里的「限速步」是同一个道理。
这套算术之所以要写成脚本,是因为它要反复重算:需求一改、有人休假、加一个任务,瓶颈可能从策划跳到服务端,人工算容易漏。
把 backlog 写成配置
下面这段是 D 层配置节选(完整文件 config/ch07_capacity.json),给出计划窗口与五个职能的产能,backlog 里每个任务是一组人天估计。注意《王朝纪元》这版刻意把策划需求堆到瓶颈位:
<!-- snippet: config/ch07_capacity.json(节选) -->
{
"plan": { "startDay": 0, "launchDay": 42, "variancePct": 40 },
"capacity": [
{ "discipline": "design", "headcount": 3, "availabilityPct": 80 },
{ "discipline": "art", "headcount": 4, "availabilityPct": 75 },
{ "discipline": "server", "headcount": 2, "availabilityPct": 70 }
],
"backlog": [
{ "id": "T-103", "title": "联盟远征玩法",
"estimates": { "design": 12, "art": 8, "client": 10, "server": 8, "qa": 6 } }
]
}
variancePct 是留给下一段蒙特卡洛的估计波动幅度;确定性排期先不看它。
确定性排期:先算点估计
下面这段是 S 层仿真脚本(完整文件 sim/ch07_capacity.py),把每个职能的完成日算出来,取瓶颈,判定是否赶得上 launchDay,并产出 JSON、燃尽 CSV 与 SVG。
# file: sim/ch07_capacity.py
# tier: S
# verify: python3 sim/ch07_capacity.py --config config/
"""内容产能与排期:人天需求 vs 每日产能,判定能否按期交付。
读 config/ch07_capacity.json。每日产能 = headcount × availabilityPct // 100(人天)。
完成日 = ceil(需求 / 每日产能),全程整数;瓶颈 = 完成日最大的职能。
"""
from __future__ import annotations
import argparse
import json
import os
import sys
import xml.etree.ElementTree as ET
HERE = os.path.dirname(os.path.abspath(__file__))
OUT = os.path.join(HERE, "..", "out")
def load_config(cfg_dir: str, name: str) -> dict:
with open(os.path.join(cfg_dir, name + ".json"), encoding="utf-8") as f:
return json.load(f)
def daily_capacity(entry: dict) -> int:
return entry["headcount"] * entry["availabilityPct"] // 100
def demand_per_discipline(cfg: dict) -> dict[str, int]:
demand: dict[str, int] = {c["discipline"]: 0 for c in cfg["capacity"]}
for task in cfg["backlog"]:
for d, md in task["estimates"].items():
demand[d] = demand.get(d, 0) + md
return demand
def plan(cfg: dict) -> dict:
launch = cfg["plan"]["launchDay"]
demand = demand_per_discipline(cfg)
rows = []
for c in cfg["capacity"]:
d = c["discipline"]
cap = daily_capacity(c)
need = demand.get(d, 0)
days = (need + cap - 1) // cap if cap > 0 else 10 ** 9
rows.append({"discipline": d, "demand": need, "dailyCapacity": cap,
"completionDay": days, "slackDays": launch - days})
bottleneck = max(rows, key=lambda r: r["completionDay"])
verdict = "OK" if bottleneck["completionDay"] <= launch else "FAIL"
return {
"launchDay": launch,
"totalManDays": sum(r["demand"] for r in rows),
"perDiscipline": rows,
"bottleneck": bottleneck["discipline"],
"bottleneckCompletionDay": bottleneck["completionDay"],
"verdict": verdict,
"shortByDays": max(0, bottleneck["completionDay"] - launch),
}
def write_csv(cfg: dict, rep: dict, path: str) -> None:
launch = rep["launchDay"]
with open(path, "w", encoding="utf-8") as f:
f.write("day," + ",".join(f"remaining_{r['discipline']}" for r in rep["perDiscipline"]) + "\n")
for day in range(0, launch + 1):
cells = []
for r in rep["perDiscipline"]:
remaining = max(0, r["demand"] - day * r["dailyCapacity"])
cells.append(str(remaining))
f.write(f"{day}," + ",".join(cells) + "\n")
def write_svg(rep: dict, path: str) -> None:
W, H = 720, 320
PAD_L, PAD_R, PAD_T, PAD_B = 56, 24, 24, 40
launch = rep["launchDay"]
bn = next(r for r in rep["perDiscipline"] if r["discipline"] == rep["bottleneck"])
hi = max(r["demand"] for r in rep["perDiscipline"])
inner_w, inner_h = W - PAD_L - PAD_R, H - PAD_T - PAD_B
def x_of(day: int) -> float:
return PAD_L + inner_w * day / launch
def y_of(v: int) -> float:
return PAD_T + inner_h * (1 - v / hi)
svg = ET.Element("svg", {"xmlns": "http://www.w3.org/2000/svg",
"viewBox": f"0 0 {W} {H}", "width": str(W), "height": str(H)})
ET.SubElement(svg, "rect", {"x": "0", "y": "0", "width": str(W), "height": str(H), "fill": "#ffffff"})
ET.SubElement(svg, "line", {"x1": str(PAD_L), "y1": str(PAD_T), "x2": str(PAD_L), "y2": str(H - PAD_B),
"stroke": "#333", "stroke-width": "1"})
ET.SubElement(svg, "line", {"x1": str(PAD_L), "y1": str(H - PAD_B), "x2": str(W - PAD_R),
"y2": str(H - PAD_B), "stroke": "#333", "stroke-width": "1"})
colors = {"design": "#1f6feb", "art": "#d29922", "client": "#3fb950",
"server": "#d1242f", "qa": "#8250df"}
for r in rep["perDiscipline"]:
pts = []
for day in range(0, launch + 1):
remaining = max(0, r["demand"] - day * r["dailyCapacity"])
pts.append((x_of(day), y_of(remaining)))
ET.SubElement(svg, "polyline", {"points": " ".join(f"{x:.1f},{y:.1f}" for x, y in pts),
"fill": "none", "stroke": colors.get(r["discipline"], "#555"),
"stroke-width": "2" if r["discipline"] == rep["bottleneck"] else "1"})
# 交付线
xl = x_of(launch)
ET.SubElement(svg, "line", {"x1": f"{xl:.1f}", "y1": str(PAD_T), "x2": f"{xl:.1f}",
"y2": str(H - PAD_B), "stroke": "#d1242f", "stroke-dasharray": "4 3"})
t = ET.SubElement(svg, "text", {"x": f"{xl - 4:.1f}", "y": str(PAD_T + 12),
"font-size": "11", "text-anchor": "end", "fill": "#d1242f"})
t.text = f"launch D{launch}"
ttl = ET.SubElement(svg, "text", {"x": str(PAD_L), "y": "16", "font-size": "13", "fill": "#111"})
ttl.text = f"内容燃尽曲线(瓶颈={rep['bottleneck']},完成 D{rep['bottleneckCompletionDay']})"
os.makedirs(os.path.dirname(path), exist_ok=True)
ET.ElementTree(svg).write(path, encoding="utf-8", xml_declaration=True)
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--config", default=os.path.join(HERE, "..", "config"))
args = ap.parse_args()
cfg = load_config(args.config, "ch07_capacity")
rep = plan(cfg)
os.makedirs(OUT, exist_ok=True)
with open(os.path.join(OUT, "ch07_capacity_report.json"), "w", encoding="utf-8") as f:
json.dump(rep, f, ensure_ascii=False, indent=2)
write_csv(cfg, rep, os.path.join(OUT, "ch07_capacity_burndown.csv"))
write_svg(rep, os.path.join(OUT, "ch07_capacity.svg"))
print(f"verdict={rep['verdict']} launch=D{rep['launchDay']} bottleneck={rep['bottleneck']} "
f"completion=D{rep['bottleneckCompletionDay']} shortBy={rep['shortByDays']} totalManDays={rep['totalManDays']}")
for r in rep["perDiscipline"]:
print(f" {r['discipline']:<7} demand={r['demand']:>3} cap/day={r['dailyCapacity']} "
f"completion=D{r['completionDay']} slack={r['slackDays']}")
return 0 if rep["verdict"] == "OK" else 2
if __name__ == "__main__":
sys.exit(main())
跑一遍,design 职能需求 84 人天、每日产能 3×80//100=2,完成日 ceil(84/2)=42,正好卡在 launchDay=42,slack=0:
$ python3 sim/ch07_capacity.py --config config/
verdict=OK launch=D42 bottleneck=design completion=D42 shortBy=0 totalManDays=324
design demand= 84 cap/day=2 completion=D42 slack=0
art demand= 96 cap/day=3 completion=D32 slack=10
client demand= 66 cap/day=2 completion=D33 slack=9
server demand= 38 cap/day=1 completion=D38 slack=4
qa demand= 40 cap/day=1 completion=D40 slack=2
判定是 OK,但 slack=0 意味着一天都不能出意外。这正是点估计最危险的地方:它把「估计值 = 真实值」当成前提。
点估计的乐观偏差:蒙特卡洛基准
人天估计从来不是确定的数。下面这段是 B 层基准脚本(完整文件 sim/ch07_capacity_bench.py),把每个任务的人天按 variancePct 做整数抖动,重复 trials 次,得到瓶颈完成日的分布,与点估计对照。它用固定种子的 LCG,结果可复现;B 层只出 JSON 与软告警,不硬失败。
# file: sim/ch07_capacity_bench.py
# tier: B
# verify: python3 sim/ch07_capacity_bench.py --config config/ --trials 2000
"""内容产能的蒙特卡洛基准(B 层,软告警不硬失败)。
把每个任务的人天按 variancePct 做整数抖动,重复 trials 次,得到瓶颈完成日的分布,
与确定性点估计对照,量化「乐观偏差」。LCG 固定种子,结果可复现。
"""
from __future__ import annotations
import argparse
import json
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from ch07_capacity import load_config, daily_capacity, demand_per_discipline, plan # noqa: E402
OUT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "out")
def lcg(seed: int):
state = seed & 0xFFFFFFFF
def nxt() -> int:
nonlocal state
state = (state * 1103515245 + 12345) % 2147483648
return state
return nxt
def percentile(sorted_vals: list[int], p: int) -> int:
if not sorted_vals:
return 0
idx = (p * (len(sorted_vals) - 1)) // 100
return sorted_vals[idx]
def run(cfg: dict, trials: int, seed: int) -> dict:
launch = cfg["plan"]["launchDay"]
var = cfg["plan"]["variancePct"]
caps = {c["discipline"]: daily_capacity(c) for c in cfg["capacity"]}
disciplines = list(caps)
rng = lcg(seed)
days = []
for _ in range(trials):
sums = {d: 0 for d in disciplines}
for task in cfg["backlog"]:
for d, md in task["estimates"].items():
jitter = rng() % (var + 1) # [0, var]
sums[d] += md * (100 + jitter) // 100
completion = max((sums[d] + caps[d] - 1) // caps[d] for d in disciplines)
days.append(completion)
days.sort()
det = plan(cfg)
miss = sum(1 for d in days if d > launch)
return {
"trials": trials, "seed": seed, "variancePct": var,
"deterministicCompletionDay": det["bottleneckCompletionDay"],
"p10": percentile(days, 10), "p50": percentile(days, 50),
"p80": percentile(days, 80), "p95": percentile(days, 95),
"meanDay": sum(days) // len(days),
"minDay": days[0], "maxDay": days[-1],
"biasDays": percentile(days, 50) - det["bottleneckCompletionDay"],
"missProbabilityPct": miss * 100 // trials,
"launchDay": launch,
}
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--config", default=os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "config"))
ap.add_argument("--trials", type=int, default=2000)
ap.add_argument("--seed", type=int, default=20261012)
args = ap.parse_args()
cfg = load_config(args.config, "ch07_capacity")
rep = run(cfg, args.trials, args.seed)
os.makedirs(OUT, exist_ok=True)
with open(os.path.join(OUT, "ch07_capacity_bench.json"), "w", encoding="utf-8") as f:
json.dump(rep, f, ensure_ascii=False, indent=2)
print(f"trials={rep['trials']} seed={rep['seed']} variance={rep['variancePct']}%")
print(f"deterministic=D{rep['deterministicCompletionDay']} p10=D{rep['p10']} p50=D{rep['p50']} "
f"p80=D{rep['p80']} p95=D{rep['p95']} mean=D{rep['meanDay']}")
print(f"bias(P50-det)=+{rep['biasDays']}d P(overrun>D{rep['launchDay']})={rep['missProbabilityPct']}%")
if rep["biasDays"] > 0:
print(f"WARN 点估计乐观 {rep['biasDays']} 天:确定性排期未计入估计方差")
return 0
if __name__ == "__main__":
sys.exit(main())
2000 次试验的输出(同机相对量级):
$ python3 sim/ch07_capacity_bench.py --config config/ --trials 2000
trials=2000 seed=20261012 variance=40%
deterministic=D42 p10=D46 p50=D48 p80=D50 p95=D51 mean=D48
bias(P50-det)=+6d P(overrun>D42)=100%
WARN 点估计乐观 6 天:确定性排期未计入估计方差
结论很刺眼:确定性排期说「刚好赶上 D42」,蒙特卡洛却说中位数完成日在 D48,P(超期)=100%。原因是双重的——每个任务的人天被向上抖动,而瓶颈又是若干职能的最大值,max 运算对右尾尤其敏感。这不是模型太悲观,而是点估计太乐观:它把一堆右偏分布各自取均值再相加,系统性低估。
从点估计到区间:排期要给区间
实务上的动作有三个:
- 报区间而非点:对外承诺用
P80(此例D50),对内排期盯P50(D48),别再拿D42当计划。 - 留缓冲:
launchDay与P80之间的 8 天就是该版本的内容缓冲;缓冲不是浪费,是吸收方差的成本。 - 砍需求或加产能:若不能推迟上线,就回到
backlog砍任务或提availabilityPct,重跑脚本看瓶颈是否移位。这与 7.1 的冲突修复是同一种「改配置、重跑、看报告」的闭环。
产能模型的下一步是把它接到闭环经济仿真上(12 章),因为内容上线会改变产出与消耗;本节只负责「人够不够」。
阅读导航:上一节:7.1 版本列车与赛季结构 · 下一节:7.3 灰度、放量与回滚 。
继续阅读
探索更多技术文章
浏览归档,发现更多关于系统设计、工具链和工程实践的内容。