第 142 章 聊天服务：实现

本章导读

第 141 章定案六个设计决策（房间模型/线协议/定序/状态机/存储/降级）；本章是 chatd 的实现章——与第 140 章（configd 实现）同构的方法论：决策逐条落地、验收逐条对账、修正留痕。实现的地形：实体层（用户/房间/消息环三个结构体与其容器——决策一的落地）；协议层（行定界+JSON 的收帧/分派/回帧——决策二）；会话层（握手→进房→发言→退房的状态机——决策四）；定序与补齐（取号/广播/游标/环回绕——决策三+五）；收尾与降级（终态保证/踢出/拒绝——决策四+六）。本章压轴是压测审计——100 虚拟用户×10 房间的序号全房一致性证明（验收 ⑧ 的机器形态）——设计章"序号无洞"的语义在并发下被逐帧审计。

142.1引入

实现从帧字典开始（不是从 socket 开始）——第 141 章练习 4 的帧型字典是协议权威：上行七型（握手/join/leave/say/dm/who/quit）下行九型（ok/err/broadcast/dm/history_tail/too_far/you_left/member_list/unknown）。先写字典再写分派让协议在第一天就是可查阅的——每个 on_frame 分支对应字典一行，评审对照着看。实体的骨架先于网络：用户/房间结构体与容器操作（加入/移除/查找）是纯数据逻辑——在没有任何 socket 时就能单测（第 133 章的分层测试红利）；网络层（Engine/Listener/Stream）最后接上——那时业务逻辑已经被测试保护。这个顺序与 configd 的"自底向上"一致——纯数据层先行是两个实现章共同的节奏。

142.2概念

实现的模块地图

chatd 实现的分层地图（与第 140 章的边界图同构）：

五百行实现一个带定序保证与补齐协议的聊天服务——与 configd（280 行）、webserv（400 行内）互为印证：XRT 项目代码量与业务复杂度成正比（chatd 的实体+协议+定序是三块真业务）——网络与并发的系统复杂度全部被库层吸收。这个数字也是第 145 章"装配不是写"的最后一组证据。

实体层：三个结构与容器

struct chuser {
	char Name[32];            /* 身份（独立于连接——决策一） */
	xnetstream* Stream;        /* 可空=离线 */
	uint64 Cursor;            /* 已见最大序号 */
	struct chroom* Room;      /* 当前房间（可空=大厅） */
};
struct chroom {
	char Name[32];
	uint64 NextSeq;           /* 定序计数器（决策三） */
	struct chmsg Ring[1024];  /* 消息环（决策五） */
	size_t RingHead;          /* 最老位置 */
	void* Members;            /* 成员表（第 18 章 map） */
};
struct chmsg {
	uint64 Seq;
	char From[32];
	char Text[256];
};

容器选型：成员表 成员表 map（按名查找——who/join 的热路径，第 18 章）；房间表全局一个 map（同族）。环的实现：数组+头指针（Ring[RingHead] 最老）——覆盖时头指针前移；补齐从游标到环尾按序重放。这三个结构是 chatd 的全部状态——没有任何隐藏副本（连接上的 SetData 挂用户指针——第 67 章上下文机制）。

协议层的帧生成侧

收帧之外还有回帧（下行九型）——生成的形态：JSON Writer（第 32 章）逐帧产一行。帧模板函数（每型一个小函数——broadcast/dm/err/ok 共用骨架）：填 op、填载荷字段、（广播族）填 seq、Writer Finish 产合法 JSON——原子性由 Writer 保证（半帧不出门）。发送路径：帧文本→xrtNetStreamSend（行缓冲+\n）——与收帧的行定界对称（同一协议的两端用同一对原语）。错误帧的统一口径：{"op":"err","code":N,"why":"..."}——code 是枚举（BADJSON/NOOP/NICK_TAKEN/ROOM_FULL/NOT_IN_ROOM/...）、why 给人——机器判 code、人读 why（第 4 章错误模型的帧版）。

协议层：收帧与分派

收帧引擎 = 行分帧器（第 72 章）+ JSON 解析（第 32 章）：

/* 连接的可读事件：行到→JSON→按 op 分派 */
static void chat_on_data(xnetstream* Stream, ptr Ctx)
{
	chuser* U = ctx_user(Ctx);
	xlineview Line;
	while ( xrtLineReaderNext(U->Lines, &Line) ==
		XLINE_NEXT_LINE ) {
		xvalue* Msg = NULL;
		xvalue* Op = NULL;
		xstrview Op;
		if ( (Msg = chat_parse_json(Line)) == NULL ) {
			chat_send_err(U, CH_E_BADJSON);
			continue;            /* 坏帧回错误不断连 */
		}
		Op = xrtValueObjectGet(Msg, XRT_STR_LITERAL("op"));
		if ( (Op == NULL) ||
			!xrtValueGetString(Op, &Op) ) {
			chat_send_err(U, CH_E_NOOP);
		} else if ( xrtStrEqual(Op, XRT_STR_LITERAL("hello")) ) {
			chat_do_hello(U, Msg);
		} else if ( xrtStrEqual(Op, XRT_STR_LITERAL("join")) ) {
			chat_do_join(U, Msg);
		} /* ... say/dm/who/quit 同构 ... */
		xrtValueRelease(Msg);
	}
}

宽容与严格的分界（第 141 章协议决策）：行层宽容（坏行不影响连接）、JSON 层半宽容（坏 JSON 回错但不断）、业务层严格（昵称冲突/房满拒绝——但也是协议帧不是断连）。分派的形状与 pushd 的 on_message 同构——两个项目在协议层共享同一模式（学一次用两次——项目序列的复利）。

定序与广播：取号-投递-游标

static void chat_do_say(chuser* U, xvalue* Msg)
{
	chroom* R = U->Room;
	chmsg* M;
	if ( R == NULL ) { chat_send_err(U, CH_E_NOT_IN_ROOM); return; }
	/* 取号：锁内单点定序（决策三） */
	lock(R);
	M = &R->Ring[R->RingTail];
	M->Seq = ++R->NextSeq;
	copy_from(M, U->Name, Msg);
	R->RingTail = (R->RingTail + 1) % 1024;
	unlock(R);
	/* 投递：带号广播 + 本人游标推进 */
	chat_broadcast(R, M);
	U->Cursor = M->Seq;
}

广播的序保证：帧带 Seq——成员的游标在收到时推进；补齐路径（重连）：hello 带 cursor → 房间内从 cursor+1 重放到环尾 → 全部补发后回 ok；游标比环头还老 → too_far（客户端全量刷新）。环回绕的序号连续性：覆盖最老帧但 NextSeq 永远单调——补齐窗口=环容量（1024 帧内的断线可补——设计容量假设的兑现）。

状态机的三个迁移细节

实现状态机时最容易含糊的三个迁移：握手失败→关闭（拒绝帧发出后主动 Close——不等客户端自己断：服务端对协议错误的明确态度；关闭前的发送队列排空由 Stream 语义保证——第 67 章）；leave 后再 join 同房（用户实体复用——游标延续（决策一的又一次兑现：退房不清游标——重进时从历史尾巴续看））；quit 帧与 TCP 断的等价处理（优雅退出与异常断开走同一个清理路径——只差是否发 you_left——一条清理路径两种入口是防状态分叉的标准手法）。

会话状态机与终态

握手（昵称合法且不冲突→注册用户；冲突→err+关闭）：join/leave 的顺序敏感（第 141 章坑 3 的实现）：先移除成员→给本人 you_left（最后一帧）→房间广播 member_left。连接断开（TCP 层 Close 事件）：用户进离线态（Stream=NULL、游标保留）——不是销毁（重连补齐的资源）；离线超时（可配）才真清理。服务端踢（慢成员——决策六）：发送持续 AGAIN 超阈值→踢出帧→Close——序号无洞语义禁止"跳过"档（与 pushd 的关键差异——第 141 章分工对照的实现面）。

压测的执行细节

100 替身×10 房的压测在实现侧的具体形态：替身进程内驱动（第 59 章 executor 的 100 任务——每替身是状态机循环：随机动作（发言 70%/听 20%/私聊 10%）+ 帧间隔（泊松分布模拟真人节奏——第 11 章随机）+ 审计器（收帧验 Seq==Cursor+1））；服务端同进程或子进程（本地回环——网络真实但可控）；注入器（指定替身在指定帧号断线/变慢——服务端无感知的故障注入）；判定（10 分钟后：全部审计零违例 + 补齐衔接正确 + 慢踢计数与注入数一致 + stats 零泄漏）。这个拓扑的可信度论证：回环 TCP 与真实网络的差异（延迟/丢包）不影响正确性验证（序号审计是逻辑性质不是时序性质）——性能结论才需要真实网络（那是部署后的事——第 136 章的基线对比）。正确性压测在 CI、性能压测在环境——两类度量的归属分离。

复盘：实现章的完成态

chatd 实现完成——项目序列的教学意图在此收口（webserv 在第 145 章已做全书收官——本章补齐的是序列中缺的一环）：六项目的约束阶梯完整呈现——数据规模（logstat）→ 生命周期（configd）→ 连接身份与定序（chatd）→ 远程不确定性（xdl）→ 扇出与终态（pushd）→ 全栈综合（webserv）。chatd 的独特贡献是"定序"——其他项目的数据要么无序（工具）要么单写者（配置）要么可丢（推送）——只有聊天把多方并发写入的全房一致顺序作为正确性核心——它是最接近"数据库"复杂度的项目（单点定序器=微型 WAL、环=环形日志、补齐=日志回放）。二十条工程常识在 chatd 的检验：全部适用、两条（终态对账/扇出先设计）再次成为主角——常识清单的稳定性在第六次重放中确立。

压测审计：验收 ⑧ 的机器形态

审计器的位置：每个虚拟客户端（第 141 章替身设计）收帧时自带检查——Seq == Cursor+1（逐帧严格递增）——违反即报告（哪个房间哪个位置断了洞）。压测拓扑：100 替身分布 10 房、随机发言 10 分钟、注入 10 个断线点、5 个慢客户端。审计输出：全房序号链完整（无洞）、补齐后游标衔接、慢客户端被踢后其余房间不受影响。这个审计是设计章"序号无洞"语义的终极证明——不是分析而是枚举式的运行时检查（每帧都被验证）——第 133 章测试哲学的并发应用。

与设计章的对账

八条验收逐条：①六 UC 正反（帧字典+分派层的正反用例全绿）；②定序一致（压测审计——全房单调）；③补齐正确（断线注入测试：重连后游标衔接不重不漏）；④终态干净（退房/断开后的幽灵帧测试——零幽灵）；⑤慢成员隔离（慢注入：其余成员序号不受影响）；⑥房满/昵称拒绝（容量测试：拒绝帧带原因）；⑦OOM 零残留（逐点：任一点失败用户/房间状态不变）；⑧压测稳定（100×10 分钟审计全绿）。八条全过——chatd 可交付。

142.3示例

第一个完整程序：收帧引擎的原子

下面的程序来自 examples/io/line——chatd 收帧引擎（行定界层）的教学原子：

 */

#include <stdio.h>

#include <xrt.h>



int main(void)
{
	/* 混合行尾 + 末行无换行的日志样本。 */
	static const char sLog[] =
		"INFO server started\r\n"
		"WARN queue is busy\n"
		"ERROR request failed";
	xreader* pReader;
	xlinereader* pLines;
	xlineview Line;          /* 借用：到下一次 Next 前有效 */
	xlinenext Next;
	size_t iCount = 0u;
	int iResult = 1;

	



	 */
	pReader = xrtReaderFromMemory(XRT_BYTES_LITERAL(sLog));
	pLines = xrtLineReaderTake(&pReader, 1024u);
	if ( pLines == NULL ) {
		xrtReaderDestroy(pReader);
		return 1;
	}

	/* 逐行消费：每行是借用视图，处理完再取下一行。 */
	while ( (Next = xrtLineReaderNext(pLines, &Line)) == XLINE_NEXT_LINE ) {
		printf("%zu: %.*s\n", ++iCount, (int)Line.Text.Size, Line.Text.Data);
	}
	if ( Next == XLINE_NEXT_END ) {
		iResult = 0;
	}

	/* 接管模式：一个 Destroy 连底层 Reader 一起关闭。 */
	if ( !xrtLineReaderDestroy(pLines) ) {
		iResult = 1;
	}
	return iResult;
}

刚才发生了什么。 ① ReaderFromMemory→LineReaderTake 的两步与 Next 三态——chatd 的 chat_on_data 主体（把 Reader 换成连接的接收缓冲——第 67 章 Stream 的数据事件就是 Reader 的网络形态）。② 视图借用（每行即取即用）——分派完即释放、不缓存——第 141 章"帧不过夜"纪律。③ 这个原子在三个项目（137/139/141）反复出现——地基原子的复用是分层设计的直接收益。

第二个完整程序：广播帧的生成

第二个程序来自 examples/logging/file_json——JSON 帧生成的参照（每行一个 JSON 的产出形态）：

 */

#include <xrt.h>

#include <stdio.h>



int main(void)
{
	xlogfileoptions Options;
	xlogger* pLogger;
	xlogfield Field;

	if ( !xrtLogFileOptionsInit(&Options, "example_logger_json.log") ) {
		return 1;
	}
	Options.MaxBytes = 1024u * 1024u;   /* 1MB 轮转 */
	Options.BackupCount = 3u;

	pLogger = xrtLogCreate(XRT_STR_LITERAL("example"), XLOG_INFO);
	Field = xrtLogFieldUInt(XRT_STR_LITERAL("request_id"), 42u);
	if (
		(pLogger == NULL) ||
		!xrtLogAddJsonFile(pLogger, &Options, NULL) ||   /* 一行挂载 */
		(xrtLogFields(
			pLogger,
			XLOG_INFO,
			XRT_STR_LITERAL("JSON file sink ready"),
			&Field,
			1u
		) != XLOG_RESULT_WRITTEN)
	) {
		xrtLogFree(pLogger);
		return 2;
	}
	xrtLogFree(pLogger);
	printf("wrote example_logger_json.log\n");
	return 0;
}

刚才发生了什么。 ① JSON Lines 的产出形态（每行一个完整对象）与 chatd 的下行帧同构——broadcast/dm/err 每帧一个 JSON 行。② 日志的 fields 机制（结构化字段随记录）对应聊天帧的载荷字段——同一序列化需求在不同系统（日志是给机器的帧、聊天是给客户端的帧）。③ chatd 的帧生成用第 32 章 Writer 直接产 JSON（{"op":"broadcast","seq":N,...}）——这个示例是它的日志侧亲戚。

与设计章的修正记录

实现期对第 141 章设计的一处修正与两处细化（决策可修订条款的兑现）：修正——补齐的锁策略从"补齐期间持房间锁"改为"锁内快照+锁外重放"（设计未细化锁范围——实现发现持锁重放会阻塞整房取号——快照模式两全）；细化一——慢成员的阈值从"持续 AGAIN"细化为"连续 N 次投递 AGAIN 且超时 T"（可配置的量化——压测标定的产物）；细化二——离线超时从"可配"细化为默认 300 秒（重连窗口的产品假设——IM 惯例的合理默认）。三处都不动核心决策（房间模型/定序/序号无洞）——设计的外围参数在实现中被现实校准——这正是"决策与参数分层"的价值：核心稳定、外围可调。

实现走查：补齐协议的源码

补齐（重连游标衔接）是 chatd 最微妙的路径——关键段走查：

/* 重连握手：带 cursor → 快照重放 → 恢复实时 */
static void chat_do_resume(chuser* U, chroom* R, uint64 Cursor)
{
	chmsg Pending[1024];
	size_t n = 0;

	if ( R == NULL ) { chat_send_err(U, CH_E_NOT_IN_ROOM); return; }
	lock(R);
	if ( (R->NextSeq - Cursor) > 1024u ) {
		unlock(R);
		chat_send(U, "{"op":"too_far"}");
		return;                 /* 比环头还老：全量刷新 */
	}
	/* 锁内快照：Cursor+1..NextSeq 的帧拷出 */
	{
		uint64 s;
		for ( s = Cursor + 1u; s <= R->NextSeq; s++ ) {
			Pending[n++] = ring_at(R, s);
		}
	}
	unlock(R);                  /* 快照完即放锁 */
	/* 锁外重放：慢客户端不阻塞房间 */
	{
		size_t i;
		for ( i = 0; i < n; i++ ) {
			chat_send_frame(U, &Pending[i]);
		}
	}
	U->Cursor = R->NextSeq;    /* 游标衔接 */
	chat_send_ok(U);
}

走查要点。 ① (NextSeq - Cursor) > 1024 的溢出安全：NextSeq 只增不减、Cursor ≤ NextSeq 恒成立（游标只会指向已见帧）——差值非负。② 快照数组在栈上限外（1024×300 字节≈300 KB——分配在堆或按需分批）——教学形态示意的只是逻辑；真实实现分批重放（每批 64 帧防栈/内存尖峰）。③ 游标推进在重放之后——中途再断线，下次补齐从头（幂等——补齐不产生副作用）。④ ring_at(R, s) 按序号取帧（环的随机访问——环形索引换算）。

测试资产清单

chatd 的四象限测试（与第 140 章同构的交付定义）：正向（test_chat.c——六 UC 各正例+帧字典十六型各一）；竞态（test_chat_threads.c——并发发言定序审计（多线程压 1000 帧后全房序号链单调）+ 补齐交错（断线重连期间持续发言）+ 幽灵帧检查（退房后零帧））；负向（昵称冲突/房满/坏 JSON/未知 op/缺字段——各回 err 且连接保持）；OOM（test_chat_oom.c——实体操作逐点：失败时房间/用户状态不变）。替身库（test_chat_client.c——虚拟用户驱动 + 序号审计器内建）与被测共享协议代码——无第二实现。全部挂清单 tests 字段——build.py --suite chatd 双轨。

142.4契约

• 实体三结构：用户（身份+连接可空+游标）/房间（定序器+环+成员表）/消息（序号+来源+文本）——全部状态无副本。
• 容器：成员表与房间表 map（按名查找）——第 18 章。
• 收帧引擎：行定界（三态循环）+JSON（ObjectGet 取 op）——宽容与严格的分界在第 141 章。
• 分派形状：op 字符串路由——与 pushd 同构（模式复用）。
• 取号：锁内 ++NextSeq 单点定序；帧带 Seq 号；成员游标随收见推进。
• 补齐：hello 带 cursor→锁内快照锁外重放；比环头老→too_far 全量刷新。
• 环语义：1024 帧/房；覆盖最老但序号永单调；补齐窗口=环容量（断线超窗即 too_far）。
• join/leave 顺序：先移除→本人 you_left→房间广播——零幽灵帧。
• 断开语义：进离线态（游标保留）；超时才清理——身份长于连接。
• 慢成员：踢出（无跳过档）——序号无洞禁跳。

交付形态与运维

chatd 的交付（与 pushd/webserv 同族的服务形态）：独立进程（静态库或单头链接——监听地址/房容量/环长/慢阈值/离线超时从环境读——五个参数覆盖全部可调面）；运维接口（管理 TCP 命令或本地信号：房间列表/成员计数/序号水位/连接数——运维端点的极简形态）；可观测（每房序号速率/连接计数/踢出计数周期打点——JSON Lines 同日志流）；重启语义（内存态服务——重启丢房间（教学边界）；客户端重连+全量刷新即恢复——无状态的客户端是服务重启容忍度的来源）。与 pushd 的运维面对照：同构（计数+列表+踢出）——服务类项目的运维面收敛成一个模式——第 144 章经验的又一次兑现。

环容量的产品权衡

环长 1024 的产品语义值得最后审视：它决定补齐窗口（断线超过 1024 帧即 too_far）——活跃房间（100 msg/s）的窗口只有 10 秒——弱网用户频繁全量刷新。调整空间：环长↑（内存↑：每帧 300B×1024=300KB/房——10 倍环=3MB/房×100 房=300MB——不可行）；落盘历史（决策五的演进点：环+磁盘两层——补齐先内存后磁盘——容量问题变 IO 问题）；快照+增量（全量刷新时发压缩快照而非逐帧——刷新成本降低）。三个方向的共同教训：内存环是教学形态的正确选择、生产形态的起点而非终点——边界声明（第 141 章）的诚实在此兑现：设计时就说清了它不是生产答案。

常见错误

lock(R);
seq = ++R->NextSeq;
store(R, seq, msg);
broadcast(R, seq, msg);   /* 锁内发：慢者拖全体 */
unlock(R);

lock(R);
seq = ++R->NextSeq;
store(R, seq, msg);
unlock(R);
broadcast(R, seq, msg);   /* 锁外发：取号不被阻塞 */

replay_from(cursor);       /* 补齐开始 */
/* 新广播同时在到——交错 */

lock(R);
snapshot_ring(R, cursor, &pending);   /* 锁内快照待补帧 */
unlock(R);
send_pending(U, &pending);            /* 补完 */
resume_live(U);                        /* 再接收实时 */

on_close(U) { destroy_user(U); }    /* 断开即销毁：游标/房间关系全丢 */

on_close(U) {
	U->Stream = NULL;    /* 离线：连接空、身份与游标保留 */
	schedule_timeout(U);  /* 超时才真清理 */
}

/* 投递：房间全体（锁外）+ 各自游标推进 */
static void chat_broadcast(chroom* R, const chmsg* M)
{
	size_t n = member_snapshot(R, NULL, 0);   /* 数成员 */
	chuser** Members = malloc(n * sizeof(void*));
	if ( Members == NULL ) { return; }        
		重连补齐兜底（序号不丢） */
	member_snapshot(R, Members, n);           /* 稳定快照 */
	for ( size_t i = 0; i < n; i++ ) {
		chuser* U = Members[i];
		if ( (U->Stream == NULL) ||
			!chat_send_frame(U, M) ) {
			/* 离线：靠游标+补齐；在线发送失败：记慢 */
			mark_slow(U);
			continue;
		}
		U->Cursor = M->Seq;    /* 本人收见推进 */
	}
	free(Members);
}

编程练习

小结