上一节把「功能归谁」钉死了，但一个功能是不是稳定，取决于它背后那份状态是不是稳定。SLG 里最典型的翻车方式，不是某个功能没人做，而是同一份数据被两个模块同时改：城建在扣木材，联盟互助也在扣木材，两边各算各的，最后资源凭空多出来或者凭空消失。
先划边界。站内的 SLG 技术架构专题 讲的是技术架构：前后端分层、服务端模块拆分、通信协议，明确写了「不讨论玩法数值」。本节讲的是它没碰的另一半——产品侧的状态归属，也就是「这份数据由哪个模块负责写、哪些模块只能读」。至于账本层面的守恒校验（每一笔增减都要对得上），那是第 9 章的 账本模型与守恒 的事；本节只解决「谁有权写」。
状态要认主：单写者原则
给系统上锁之前，先定一条规矩：每份状态有且只有一个写者（single writer）。
- 「写」= 改变这份状态的值；
- 「读」= 只获取这份状态的值，不改变它；
- 一份状态可以被多个模块读，但只能被一个模块写。
这条规矩的价值在于把「改数据」这件事收敛到一个点上。只要所有人都知道「资源存量只有城建能写」，那么：
- 排查 bug 时，你只需要看一个模块的写入逻辑；
- 改数值时，你只需要改一处；
- 做存档时，你只需要保证一个模块的写入是原子的。
反过来，一旦一份状态有两个写者，你就失去了这三个性质——而这正是大型 SLG 项目里最常见、也最难查的一类问题。
写者、读者与流
把每份状态写成一行，至少要有四个字段：owner（归属模块）、writers（写者）、readers（读者）、id（状态标识）。再补一组 flows，描述「谁把哪份状态的变化推给谁」——注意这是事件推送，不是数据所有权转移。
这里有个容易混的点：读者多不等于耦合重。一份状态被十个模块读，只要写者唯一，它就是一份稳定的「只读资源」；一份状态被两个模块写，哪怕只有两个，它也是一个随时会炸的定时炸弹。判断耦合要看写，不是看读。这一点会在下面的基准里体现。
把状态归属写成配置
下面这段是 D 层配置：9 份状态的归属表，外加 6 条跨模块事件流。注意每份状态的 writers 长度恒为 1，且等于 owner。
{
"version": 1,
"note": "状态归属表：每份状态（state）有且只有一个写者（writers 长度恒为 1 且等于 owner），可被多个模块读（readers）。flows 描述模块间的事件推送（谁把哪份状态的变化推给谁）。",
"modules": ["city", "army", "map", "alliance", "quest", "shop", "event", "account"],
"states": [
{ "id": "city.resources", "name": "资源存量", "owner": "city", "writers": ["city"], "readers": ["city", "army", "shop", "event"] },
{ "id": "city.buildings", "name": "建筑等级", "owner": "city", "writers": ["city"], "readers": ["city", "army", "quest"] },
{ "id": "army.troops", "name": "部队编成", "owner": "army", "writers": ["army"], "readers": ["army", "map"] },
{ "id": "map.tiles", "name": "地块归属", "owner": "map", "writers": ["map"], "readers": ["map", "army", "alliance"] },
{ "id": "alliance.members", "name": "联盟成员", "owner": "alliance", "writers": ["alliance"], "readers": ["alliance", "event"] },
{ "id": "quest.progress", "name": "任务进度", "owner": "quest", "writers": ["quest"], "readers": ["quest", "event"] },
{ "id": "shop.wallet", "name": "付费钱包", "owner": "shop", "writers": ["shop"], "readers": ["shop", "event"] },
{ "id": "event.season", "name": "赛季进度", "owner": "event", "writers": ["event"], "readers": ["event", "quest"] },
{ "id": "account.profile", "name": "账号档案", "owner": "account", "writers": ["account"], "readers": ["account", "city", "alliance", "shop"] }
],
"flows": [
{ "id": "city.produced", "from": "city", "to": "army", "payload": "city.resources", "kind": "event" },
{ "id": "army.marched", "from": "army", "to": "map", "payload": "army.troops", "kind": "event" },
{ "id": "map.tileChanged", "from": "map", "to": "alliance", "payload": "map.tiles", "kind": "event" },
{ "id": "alliance.memberChanged", "from": "alliance", "to": "event", "payload": "alliance.members", "kind": "event" },
{ "id": "quest.completed", "from": "quest", "to": "event", "payload": "quest.progress", "kind": "event" },
{ "id": "shop.purchased", "from": "shop", "to": "event", "payload": "shop.wallet", "kind": "event" }
]
}
readers 里包含 owner 自己——一个模块当然可以读自己写的状态。真正的边界信息在「除了自己还有谁」。
校验它
下面这段是 S 层仿真脚本，把归属表逐条校验：单写者、归属一致、引用有效、无死状态、模块覆盖，产出 JSON、CSV 与一张模块-状态二部图（写实线、读虚线）。
# file: sim/ch05_state_ownership.py
# tier: S
# verify: python3 sim/ch05_state_ownership.py --config config/
"""状态归属校验（S 层）：单写者、归属一致、引用有效、无死状态、模块覆盖。
产出 JSON（stdout + out/）+ CSV（各模块写/读状态数）+ SVG（模块-状态二部图）。
"""
from __future__ import annotations
import argparse
import csv
import json
import os
import xml.etree.ElementTree as ET
NS = "http://www.w3.org/2000/svg"
def load(cfg_dir: str, name: str) -> dict:
with open(os.path.join(cfg_dir, f"{name}.json"), encoding="utf-8") as f:
return json.load(f)
def validate(cfg: dict) -> dict:
modules, states, flows = list(cfg["modules"]), cfg["states"], cfg["flows"]
modset, V = set(modules), []
for s in states:
sid, writers, readers = s["id"], s.get("writers", []), s.get("readers", [])
if len(writers) != 1: # 单写者：同一份状态只能有一个写模块
V.append({"code": "MULTI_WRITER", "target": sid, "detail": f"写者 {len(writers)} 个：{writers}"})
if s.get("owner") not in modset:
V.append({"code": "BAD_OWNER", "target": sid, "detail": f"归属模块不存在：{s.get('owner')}"})
elif writers and s["owner"] not in writers:
V.append({"code": "OWNER_NOT_WRITER", "target": sid, "detail": f"归属 {s['owner']} 不在写者 {writers}"})
for m in writers + readers:
if m not in modset:
V.append({"code": "BAD_MODULE_REF", "target": sid, "detail": f"引用未知模块：{m}"})
if not readers:
V.append({"code": "DEAD_STATE", "target": sid, "detail": "无任何模块读取（死状态）"})
ids = {s["id"] for s in states}
for fl in flows:
if fl["from"] not in modset or fl["to"] not in modset:
V.append({"code": "BAD_FLOW_MODULE", "target": fl["id"], "detail": f"端点模块不存在：{fl['from']}→{fl['to']}"})
if fl["payload"] not in ids:
V.append({"code": "BAD_FLOW_PAYLOAD", "target": fl["id"], "detail": f"载荷状态不存在：{fl['payload']}"})
involved = set()
for s in states:
involved.update(s.get("writers", []))
involved.update(s.get("readers", []))
for m in modules:
if m not in involved:
V.append({"code": "IDLE_MODULE", "target": m, "detail": "未参与任何状态"})
per = [{"module": m, "writes": sum(1 for s in states if m in s.get("writers", [])),
"reads": sum(1 for s in states if m in s.get("readers", [])),
"writeStates": [s["id"] for s in states if m in s.get("writers", [])],
"readStates": [s["id"] for s in states if m in s.get("readers", [])]} for m in modules]
return {"version": 1, "ok": not V,
"counts": {"modules": len(modules), "states": len(states), "flows": len(flows),
"writeEdges": sum(len(s.get("writers", [])) for s in states),
"readEdges": sum(len(s.get("readers", [])) for s in states), "violations": len(V)},
"violations": V, "perModule": per}
def write_svg(cfg: dict, report: dict, path: str) -> None:
modules, states = list(cfg["modules"]), cfg["states"]
rh, pad, top, mx, sx = 32, 30, 56, 180, 660
H = top + max(len(modules), len(states)) * rh + pad
W = sx + 250 + pad
ET.register_namespace("", NS)
svg = ET.Element(f"{{{NS}}}svg", {"width": str(W), "height": str(H), "viewBox": f"0 0 {W} {H}"})
ET.SubElement(svg, f"{{{NS}}}rect", {"width": str(W), "height": str(H), "fill": "#0f1720"})
ET.SubElement(svg, f"{{{NS}}}text", {"x": str(pad), "y": "28", "fill": "#e2e8f0", "font-size": "16",
"font-family": "sans-serif"}).text = \
f"《王朝纪元》状态归属：{len(modules)} 模块 / {len(states)} 状态 / 写 {report['counts']['writeEdges']}" \
f" · 读 {report['counts']['readEdges']} / 违规 {report['counts']['violations']}"
my = {m: top + i * rh for i, m in enumerate(modules)}
sy = {s["id"]: top + i * rh for i, s in enumerate(states)}
for m in modules:
y = my[m]
ET.SubElement(svg, f"{{{NS}}}rect", {"x": str(mx), "y": str(y - 11), "width": "110", "height": "22",
"rx": "6", "fill": "#1d4ed8"})
ET.SubElement(svg, f"{{{NS}}}text", {"x": str(mx + 55), "y": str(y + 4), "fill": "#fff", "font-size": "12",
"text-anchor": "middle", "font-family": "sans-serif"}).text = m
for s in states:
y = sy[s["id"]]
ET.SubElement(svg, f"{{{NS}}}rect", {"x": str(sx), "y": str(y - 11), "width": "230", "height": "22",
"rx": "6", "fill": "#334155"})
ET.SubElement(svg, f"{{{NS}}}text", {"x": str(sx + 115), "y": str(y + 4), "fill": "#e2e8f0",
"font-size": "12", "text-anchor": "middle",
"font-family": "sans-serif"}).text = f"{s['name']} · {s['id']}"
for w in s.get("writers", []):
ET.SubElement(svg, f"{{{NS}}}line", {"x1": str(mx + 110), "y1": str(my[w]), "x2": str(sx),
"y2": str(y), "stroke": "#f59e0b", "stroke-width": "2"})
for rd in s.get("readers", []):
if rd in s.get("writers", []):
continue
ET.SubElement(svg, f"{{{NS}}}line", {"x1": str(mx + 110), "y1": str(my[rd]), "x2": str(sx),
"y2": str(y), "stroke": "#0ea5e9", "stroke-width": "1",
"stroke-dasharray": "4 3"})
ET.ElementTree(svg).write(path, encoding="utf-8", xml_declaration=True)
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--config", default="config/")
ap.add_argument("--out", default=os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "out"))
args = ap.parse_args()
cfg = load(args.config, "ch05_state_ownership")
report = validate(cfg)
os.makedirs(args.out, exist_ok=True)
with open(os.path.join(args.out, "ch05_state_ownership.json"), "w", encoding="utf-8") as f:
json.dump(report, f, ensure_ascii=False, indent=2)
with open(os.path.join(args.out, "ch05_state_ownership.csv"), "w", encoding="utf-8", newline="") as f:
w = csv.DictWriter(f, fieldnames=["module", "writes", "reads", "writeStates", "readStates"])
w.writeheader()
for pm in report["perModule"]:
w.writerow({**pm, "writeStates": "|".join(pm["writeStates"]), "readStates": "|".join(pm["readStates"])})
write_svg(cfg, report, os.path.join(args.out, "ch05_state_ownership.svg"))
print(json.dumps(report, ensure_ascii=False, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())
跑一遍：9 状态 24 读边 0 违规
跑 python3 sim/ch05_state_ownership.py --config config/，报告头是：
{"version": 1, "ok": true,
"counts": {"modules": 8, "states": 9, "flows": 6, "writeEdges": 9, "readEdges": 24, "violations": 0}}
writeEdges 是 9，正好等于状态数——每份状态一个写者，没有例外；readEdges 是 24，平均每份状态被 2.7 个模块读。各模块的读写分布是：
| 模块 | 写状态数 | 读状态数 | 关键读入 |
|---|---|---|---|
| city | 2 | 3 | 账号档案 |
| army | 1 | 4 | 资源存量、建筑等级、地块归属 |
| map | 1 | 2 | 部队编成 |
| alliance | 1 | 3 | 地块归属、账号档案 |
| quest | 1 | 3 | 建筑等级、赛季进度 |
| shop | 1 | 3 | 资源存量、账号档案 |
| event | 1 | 5 | 资源存量、联盟成员、任务进度、付费钱包 |
| account | 1 | 1 | 账号档案 |
这张表立刻暴露了一个结构信号：活动模块读了 5 份状态，是全场最多的。这不是 bug，但值得警惕——一个模块读的状态越多，它就越容易在别人改状态时被动跟着改。下一节我们用基准把这个信号量化出来。
双写为什么是灾难
把「同一份状态被两个模块写」的情形单独拎出来看。假设有人给 city.resources 加了第二个写者 alliance（理由通常很正当：「联盟互助要扣资源」），会发生什么：
- 城建按自己的结算 tick 每小时加资源；
- 联盟在互助时直接减资源；
- 两个模块各自持有「资源存量」的一份缓存；
- 结算时两边用各自的缓存写回，后写的覆盖先写的——先写的那部分产出就凭空蒸发了。
更隐蔽的是，这种 bug 在单机测试里几乎测不出来：只要操作顺序固定，两边不会同时写。它只在并发或者时序交错时暴露，而 SLG 的服务器恰恰是并发环境。
所以「单写者」不是一句架构洁癖，它是一道并发安全的底线。校验器里的 MULTI_WRITER 就是这道底线的守卫：任何一份状态的 writers 长度一旦不等于 1，CI 直接红。真要做「联盟互助扣资源」，正确做法是联盟发一条命令给城建模块，由城建这个唯一写者去扣——这就是下一节 跨模块契约
要讲的东西。
耦合度基准
有了归属表，就能算每个模块的耦合度：fan-out 是它读到的外部状态所属的不同模块数，fan-in 是读到它拥有的状态的不同模块数。下面这段是 B 层基准脚本，把每个模块的 fan-out / fan-in 与一个期望上界对照，超出只软告警、不硬失败。
# file: sim/ch05_contract_bench.py
# tier: B
# verify: python3 sim/ch05_contract_bench.py --config config/
"""状态耦合度基准（B 层）：从状态归属表算出每个模块的 fan-out / fan-in，与基准上界对照。
软告警不硬失败，退出码恒为 0；产出 JSON（stdout + out/ch05_contract_bench.json）。
"""
from __future__ import annotations
import argparse
import json
import os
def load(cfg_dir: str, name: str) -> dict:
with open(os.path.join(cfg_dir, f"{name}.json"), encoding="utf-8") as f:
return json.load(f)
def coupling(ownership: dict) -> list[dict]:
states = ownership["states"]
owner = {s["id"]: s["owner"] for s in states}
rows = []
for m in ownership["modules"]:
reads = [s for s in states if m in s.get("readers", [])]
fan_out = {owner[s["id"]] for s in reads if owner[s["id"]] != m}
owned = [s["id"] for s in states if s["owner"] == m]
fan_in = {r for s in states if s["owner"] == m for r in s.get("readers", []) if r != m}
rows.append({"module": m, "fanOut": len(fan_out), "fanIn": len(fan_in)})
return rows
def bench(cfg_dir: str) -> dict:
ownership = load(cfg_dir, "ch05_state_ownership")
base = load(cfg_dir, "ch05_bench_baseline")
rows, warns = [], []
for c in coupling(ownership):
b = base["modules"].get(c["module"], base["defaults"])
for axis, lim in (("fanOut", b["maxFanOut"]), ("fanIn", b["maxFanIn"])):
if c[axis] > lim:
warns.append({"module": c["module"], "axis": axis, "actual": c[axis], "limit": lim,
"detail": f"{c['module']} 的 {axis}={c[axis]} 超出基准上界 {lim}"})
rows.append({**c, "maxFanOut": b["maxFanOut"], "maxFanIn": b["maxFanIn"],
"status": "warn" if c["fanOut"] > b["maxFanOut"] or c["fanIn"] > b["maxFanIn"] else "ok"})
return {"version": 1, "tier": "B", "note": "软告警基准：只报告不硬失败，绝对数值不可移植。",
"counts": {"modules": len(rows), "warnings": len(warns)}, "modules": rows, "warnings": warns}
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--config", default="config/")
ap.add_argument("--out", default=os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "out"))
args = ap.parse_args()
report = bench(args.config)
os.makedirs(args.out, exist_ok=True)
with open(os.path.join(args.out, "ch05_contract_bench.json"), "w", encoding="utf-8") as f:
json.dump(report, f, ensure_ascii=False, indent=2)
print(json.dumps(report, ensure_ascii=False, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())
跑 python3 sim/ch05_contract_bench.py --config config/，得到 {"counts": {"modules": 8, "warnings": 1}}，唯一的告警是：
| 模块 | fan-out | 上界 | fan-in | 上界 | 状态 |
|---|---|---|---|---|---|
| city | 1 | 1 | 4 | 5 | ok |
| army | 2 | 3 | 1 | 2 | ok |
| map | 1 | 2 | 2 | 3 | ok |
| alliance | 2 | 2 | 1 | 2 | ok |
| quest | 2 | 2 | 1 | 1 | ok |
| shop | 2 | 2 | 1 | 2 | ok |
| event | 4 | 2 | 1 | 3 | warn |
| account | 0 | 0 | 3 | 4 | ok |
event 的 fan-out 是 4（它读了城建、联盟、任务、商城四个模块的状态），超出基准上界 2。这不是说活动模块做错了，而是说它承担了过多的外部依赖：任何一个上游模块改了状态结构，活动都要跟着改。合理的收敛方向是——活动模块不要直接读四家的状态，而是订阅它们各自发出的领域事件（city.produced、alliance.memberChanged……也就是归属表里的 flows），把「主动拉」换成「被动收」。这正是把耦合从 fan-out=4 压回 2 左右的标准手法。
注意这里是软告警：脚本退出码恒为 0，告警只写进 JSON。产品模块边界允许在运营压力下临时越界，基准的作用是让越界被看见，而不是一刀切禁掉。
归属表怎么和功能树对齐
最后把两张表对齐一遍：功能树的每个 owner 都应该在归属表里作为某份状态的写者出现；归属表里的每个模块也都应该在功能树里拥有至少一个功能。两张表互相印证，模块边界才算真正闭合。
下一步是把「谁能调用谁」也定下来——状态归属解决了「谁有权写」，但还没解决「一个模块怎么请求另一个模块去写」。那就是跨模块契约：用 TypeScript 接口把命令、事件和读写权限一起写死，再用 node --test 断言「同一份状态被两个模块写」一定被检出。
阅读导航：上一节：功能树与模块职责 · 下一节：跨模块契约与接口约定 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。