第 144 章 项目五：WebSocket 实时推送服务

本章导读

卷十三项目五：实时推送服务（pushd）——WebSocket 长连接网关：客户端连上后订阅频道（{"op":"sub","ch":"metrics"}），服务端把后端产生的消息推给频道全体成员（广播）；支持踢出（kick）、频道成员查询、优雅停机（先 Seal 再清场的完整流程）。它是 xws 四章（第 110-113 章）的综合运用：路由端点（第 113 章 xrtWsServerRoute——Origin 策略与授权回调的安全入口）、连接组（第 112 章 xwsgroup——成员管理/快照广播/Seal 半关闭）、发送路径（第 111 章 writer/引用/压缩——广播的组播原语）、连接生命周期（第 110 章 pause/resume/关闭协议）。新加的系统约束是长连接并发：数百条连接同时活着的生命周期管理、广播的扇出放大、慢客户端对整体的影响——项目一二四的经验加上这条后在静态服务器（第 145 章）汇成全栈。

144.1引入

需求场景：监控仪表盘要实时显示服务指标——轮询（1 秒一拉）延迟高且浪费；SSE（第 107 章）单向够用但要客户端分频道的订阅语义与双向控制（sub/unsub/kick）——WebSocket 的双向长连接是自然形态。pushd 的接口设计：连接协议（升级到 /ws，子协议 push.v1）；订阅消息（JSON：sub/unsub 频道名）；推送消息（JSON：频道+载荷+序号——客户端按序号检漏）；管理动作（kick 指定连接、列出频道成员——运维接口）。验收标准：① 100 并发连接三频道广播全达（无漏发——序号对账）；② 慢客户端（收但不读）不拖快客户端（隔离验证）；③ kick 立即生效（后续消息不到达被踢者）；④ 优雅停机：Seal 后新连接拒绝、在途消息送达、全部连接关闭后才退出（终态计数对账）；⑤ 广播内存与连接数×消息大小解耦（引用发送——stats 验证）；⑥ 背压下服务稳定（广播不因单连接满而阻塞）。

144.2概念

架构：三层组装

订阅集的存放决策：连接组只管"连接的集合"（第 112 章边界——不管消息编码不关业务）；频道→连接集的映射是业务状态——独立的多组结构（每频道一个 xwsgroup）或 Map<频道, 组>。决策：每频道一组——组的并发语义（唯一成员/引用/快照）正是频道成员管理需要的，免费获得；备选 Map<频道, 连接数组>要自己实现全部并发语义——否决。

模块选型与理由

广播管线：从后端消息到 N 个客户端

序号的作用：每频道单调递增——客户端检测漏收（收到 3 和 5——4 丢了：主动 re-sync）。引用发送的组播升级（第 111 章）：载荷一份、N 连接的队列各持引用——释放回调计数（最后一个写完才真释放）；GroupTextAsync 内部正是这个形态。慢连接策略的三档：跳过+记慢（默认——丢一帧可接受，客户端 re-sync）；踢出（配置阈值——慢性劣化连接清理）；无界等待（不提供——单连接拖死全体违反验收 ②⑥）。

订阅协议与消息处理

收到消息（第 110 章事件）→ JSON 解析（第 32 章）→ 按字段 op 分派：sub（校验频道名合法→加入对应组 GroupAdd——首建组则创建）；unsub（GroupRemove+引用归还）；心跳（客户端 ping——连接层自动 Pong 已处理（第 110 章），应用层可选回显在线状态）。非法消息：解析失败回错误帧（JSON 错误码）不断连（协议宽容——损坏一帧不代表连接坏）；频道名非法回错误——每个错误可定位（第 4 章错误模型的应用消息版）。

优雅停机全流程

第 112 章停机模板在 pushd 的完整展开：第一步停止接受（监听层 Drain——新 HTTP 升级不再受理）；第二步 Sealing 全部频道组（GroupSeal——幂等；新订阅进不来、已有成员照常收发）；第三步终末广播（"server shutting down"帧——给客户端明确信号而非静默断）；第四步逐连接关闭（ConnClose NORMAL 码——协议化告别）；第五步等 Close 回调计数归零（全部连接终态）；第六步销毁组与路由（Release 的 Data 生命周期在此收尾——第 113 章"Router 与全部连接退出后恰好一次"）；第七步停 Engine。验收 ④ 的对账点：关闭帧送达（客户端收到了 shutting down）+ 无泄漏（组的成员计数归零）+ 退出码 0。

与前面项目经验的对位

十二条经验（第 143 章清点）在长连接形态的重放与新增：验收先行（六条——并发与时序条目占比上升）；形态决定容器（频道=组、订阅集=Map、消息=JSON——三形态三容器）；视图即取即弃（消息事件的视图即解析即分派——不留缓冲）；分层换件（换传输层 TLS/换序列化 XLON 是单层）；平台事实进设计（Origin 策略与子协议协商是 WS 的"平台事实"）；预算先算（广播扇出 = 消息率×连接数——背压与慢策略的预算依据）；测试四象限（并发测试成为主战场：100 连接×三频道×序号对账）。新增两条：扇出是放大器（1 条后端消息 × 1000 连接 = 1000 次发送——广播路径的每纳秒都被放大，引用发送不是优化而是必需）；长连接的状态清理是最终考验（短连接的清理在每次请求后自然发生；长连接的清理只在停机与踢出——泄漏更隐蔽，终态计数对账是唯一可靠的验收）。

144.3示例

第一个完整程序：频道组的成员管理

下面的程序来自 examples/websocket/group——频道成员管理的接入模板：

#include <stdio.h>

#include <xws.h>



/* Upgrade 成功后把借用连接加入应用连接组。 */
static void connectionOpen(xwsgroup* pGroup, xwsconn* pConnection)
{
	(void)xrtWsGroupAdd(pGroup, pConnection);
}



/* 唯一 Close 回调中移除连接，归还连接组持有的引用。 */
static void connectionClose(xwsgroup* pGroup, xwsconn* pConnection)
{
	(void)xrtWsGroupRemove(pGroup, pConnection);
}



/* 示例保留回调形状，实际服务器可直接把组放入路由上下文。 */
int main(void)
{
	xwsgroup* pGroup = xrtWsGroupCreate(0);

	if ( pGroup == NULL ) {
		return 1;
	}
	(void)connectionOpen;
	(void)connectionClose;
	printf("WebSocket connection group is ready\n");
	xrtWsGroupDestroy(pGroup);
	return 0;
}

刚才发生了什么。 ① connectionOpen/Close 两个函数就是 pushd 的订阅/退订内核：升级成功 GroupAdd（进频道）、唯一 Close 回调 GroupRemove（归还引用）——连接的组生命周期与连接生命周期严格同步（没有退订泄漏的可能——Close 是唯一终态保证恰好一次）。② pushd 在此之上加频道路由（多组）与消息协议——成员管理的并发语义（重复 Add 幂等/容量拒绝/快照稳定）全部由组继承。③ 真实服务器把组放进路由 Data（第 113 章）——本示例锁定的是"接入的形状"。

第二个完整程序：异步广播操作对象

第二个程序来自 examples/websocket/group_future——可等待的广播：

#include <stdio.h>

#include <xws.h>



/* 展示空组广播也会返回可等待、可检查的完整操作对象。 */
int main(void)
{
	xwsgroup* pGroup = xrtWsGroupCreate(0);
	xwsgroupop* pOperation;

	if ( pGroup == NULL ) {
		return 1;
	}
	pOperation = xrtWsGroupTextAsync(
		pGroup,
		XRT_STR_LITERAL("service-ready")
	);
	if ( (pOperation == NULL) ||
		(xrtWsGroupOpWait(pOperation) != XWAIT_OK) ) {
		xrtWsGroupOpDestroy(pOperation);
		xrtWsGroupDestroy(pGroup);
		return 1;
	}
	printf(
		"members=%zu accepted=%zu rejected=%zu\n",
		xrtWsGroupOpCount(pOperation),
		xrtWsGroupOpAccepted(pOperation),
		xrtWsGroupOpRejected(pOperation)
	);
	xrtWsGroupOpDestroy(pOperation);
	xrtWsGroupDestroy(pGroup);
	return 0;
}

刚才发生了什么。 ① xrtWsGroupTextAsync 返回操作对象——发起/等待/释放三步：pushd 的广播调用形态（生产者发完不等——定时器节奏驱动；管理接口等——kick 后确认送达）。② 空组也是完整操作（零连接=零成功——聚合语义自洽）。③ 慢连接的策略点在此：操作对象不等单个慢连接（全体受理即完成——慢者的 AGAIN 由其连接自身的背压处理）——"每连接独立受理" 正是验收 ②⑥ 的机制基础。

实现走查：广播与订阅的核心源码

pushd 的两个核心循环——订阅处理与频道广播——的源码形态（在教学示例之上给出关键段走查）：

/* 消息事件：JSON 协议分派（sub/unsub） */
static void on_message(xwsconn* Conn, const uint8_t* Data,
	size_t Size, ptr Ctx)
{
	pushd* S = Ctx;
	xvalue Msg;
	xstrview Op, Ch;

	/* 解析失败：错误帧回复，连接保持（宽容契约） */
	xvalue* V;
	if ( !xrtJsonParse(Data, Size, &Msg) ||
		((V = xrtValueObjectGet(Msg,
			XRT_STR_LITERAL("op"))) == NULL) ||
		!xrtValueGetString(V, &Op) ) {
		send_error(Conn, "bad message");
		return;
	}
	if ( ((V = xrtValueObjectGet(Msg,
			XRT_STR_LITERAL("ch"))) == NULL) ||
		!xrtValueGetString(V, &Ch) ||
		!channel_name_valid(Ch) ) {
		send_error(Conn, "bad channel");
		return;
	}

	if ( xrtStrEqual(Op, XRT_STR_LITERAL("sub")) ) {
		xwsgroup* Gr = pushd_channel(S, Ch);  /* 取或建 */
		xrtWsGroupAdd(Gr, Conn);               /* 幂等 */
		send_ok(Conn, "subscribed");
	} else if ( xrtStrEqual(Op, XRT_STR_LITERAL("unsub")) ) {
		xwsgroup* Gr = pushd_find(S, Ch);
		if ( Gr != NULL ) { xrtWsGroupRemove(Gr, Conn); }
		send_ok(Conn, "unsubscribed");
	}
}

/* 频道广播：快照 + 引用发送 + 慢连接跳过 */
static void pushd_broadcast(pushd* S, xstrview Ch, str Payload)
{
	xwsgroupsnapshot* Snap;
	size_t i;

	xwsgroup* Gr = pushd_find(S, Ch);
	if ( (Gr == NULL) || xrtWsGroupSealed(Gr) ) { return; }
	Snap = xrtWsGroupSnapshotCreate(Gr);   /* 稳定快照 */
	for ( i = 0; i < xrtWsGroupSnapshotCount(Snap); i++ ) {
		xwsconn* C = xrtWsGroupSnapshotGet(Snap, i);
		xwsgroupop* Op = xrtWsGroupTextAsync(Gr, view(Payload));
		if ( Op == NULL ) { S->SlowCount++; }  /* 跳过+记慢 */
		else { xrtWsGroupOpDestroy(Op); }
	}
	xrtWsGroupSnapshotDestroy(Snap);       /* 归还成员引用 */
}

走查要点。 ① on_message 的分派顺序：先整帧解析（坏帧回错误不断连）、再字段校验、后操作执行——每级错误独立可定位（bad message/bad channel/成功回执）。② pushd_channel 取或建组——首订阅创建组、GroupAdd 幂等（重复 sub 同频道安全——第 112 章唯一成员语义的兑现）。③ pushd_broadcast 的快照循环是坑 1/2 的正面形态：快照稳定遍历、组级异步操作、失败跳过计数——三件契约一次到位。④ 停机路径的 GroupSealed 检查：Seal 后广播自然静默（终末广播在 Seal 前发——顺序即语义）。

验收矩阵与并发测试设计

六条验收的测试矩阵（第 133 章四象限的长连接版）：正向——100 连接三频道×1000 帧广播序号对账（验收 ①）；并发竞态——广播与订阅/退订并发（快照语义的实测）、广播与 kick 并发（Remove 后不触达——验收 ③）；慢客户端——1 慢（收不读）+99 快（验收 ②：快的序号不受影响）；背压——注入生产率 > 消费率（验收 ⑥：服务稳定、慢计数上升）；停机——停机时 50 活跃连接（验收 ④：shutting down 帧全达、计数归零、退出码 0）；资源——OOM 逐点（广播路径分配失败——跳过该帧不崩）。测试的客户端侧也要并发框架——第 59 章 executor 驱动 100 个虚拟客户端（连/订/收/对账）——测试代码本身是并发的，它的正确性靠序号对账（不靠时序假设）。

与全书知识地图的对位

pushd 踩过的砖：第 4 章（错误模型——消息级错误帧）、第 18 章（Map——频道名到组的路由）、第 25 章（字符串——频道名校验与 op 比较）、第 32 章（JSON——订阅协议的编解码）、第 55-59 章（并发/executor——测试客户端框架）、第 64 章（事件端口——Engine 驱动的底层）、第 67 章（TCP——连接的传输基座）、第 94 章（HTTP 升级——WS 的握手层）、第 95-97 章（WebSocket 协议核心——帧/流/关闭）、第 104-105 章（HTTP 服务端——Router 与中间件的入口）、第 110-113 章（xws 四章——本项目的直接工具箱）。十六个章节在 pushd 里各就各位——项目五是目前覆盖面最宽的项目（静态服务器收官章会更宽）。知识地图的验证意义在于：没有一个章节的知识是孤立的——每个都在某个真实形态里被组合过。

144.4契约

• 端点安全：xrtWsServerRoute 固定端点——Origin 三档默认（SAME_HOST_OR_ABSENT）+ Authorize 可选令牌校验；Open 借用连接、长期持有须 Ref 后存（第 113 章契约）。
• 频道即组：每频道一个 xwsgroup——成员语义（幂等 Add/容量拒绝/稳定快照）由组提供；订阅集与成员管理分离（业务归应用层）。
• 广播管线：快照（稳定遍历+加引用）→ 逐成员引用发送 → 快照销毁归还引用；载荷一份内存 N 连接共享（内存与扇出解耦）。
• 慢客户端三档：跳过+记慢（默认）/踢出（阈值）/不提供无界等待——单连接不拖全体。
• 序号协议：每频道单调递增——客户端按序号检漏并主动 re-sync（重启不丢语义的运维红利）。
• 消息宽容：单帧解析失败回错误帧不断连；每个错误可定位。
• 停机七步：停接→Seal（幂等）→终末广播→逐连接 Close→计数归零→销毁组与路由→停 Engine——终态对账。
• Data 生命周期：路由 Release 在 Router 与全部连接退出后恰好一次（第 113 章契约）。
• 踢出语义：GroupRemove 先出组、再 ConnClose——出组后的广播不再触达被踢者（验收 ③ 的机制基础）。

交付形态与运维接口

pushd 的交付与运维设计：部署形态——独立进程（单头或静态库链接——第 132 章选型）；监听地址/频道容量/慢阈值从命令行或环境（第 43 章）读入——工具的配置面保持最小（三个参数够部署；复杂配置是 configd 的领域）。运维接口——管理端点（HTTP：列频道/成员数/慢计数/踢出）挂在同一 Router 的另一路径（第 105 章 HTTP 路由与 WS 端点共存——第 113 章"一个 Router 两类端点"的形态）；输出 JSON（第 32 章）供脚本消费。可观测性——连接计数/频道计数/广播速率/慢计数周期打点（第 38 章 logger 或直接 stdout JSON Lines——第 139 章 file_json 形态）；stats（第 6 章）监控内存水位。优雅重启——停机七步 + systemd 式监督：重启窗口客户端重连（第 110 章重连是客户端责任）+ 序号 re-sync 补齐丢失帧——重启不丢语义是序号协议的运维红利。

与 SSE 推送服务的边界复查

pushd 完成后值得复查第 107 章的选型判断：什么时候 SSE 够、什么时候必须 WebSocket？pushd 的三个需求里，频道订阅（客户端告诉服务端要什么）与 kick 管理（双向控制）是 SSE 做不了的——SSE 只能服务端推、客户端没有上行通道（订阅要靠 URL 参数——频道多了不现实）；序号推送本身 SSE 也能做（SSE 也有 id 字段与 Last-Event-ID 续传——第 107 章的机制）。所以分界线精确化：纯下行推送用 SSE（更简单——走 80 端口、自动重连浏览器内建）；需要客户端上行控制（订阅/确认/心跳定制）用 WebSocket。pushd 的订阅语义使它站在 WS 一侧——但如果需求退化为"全员广播同一流"（如全站公告），SSE 是更轻的答案。选型复查（每次实现完一个方案后回头看替代方案）是设计能力的组成部分——它让下一次选型更快更准。

压测方法与容量规划

第 136 章方法在长连接服务的落地：压测问题（"1000 连接×3 频道×100 帧/秒的广播下，P99 帧延迟与内存水位是多少？"——先写问题再搭实验）；负载模型——虚拟客户端（executor 驱动：连接/订阅/收帧/对账四阶段）+ 生产者（定时器定速率）——两端都是确定性脚本（无真网络依赖，本地回环——第 104 章 mock 思想的压测版）；度量——帧延迟分布（发送到接收的延时直方图——第 42 章时间戳打点）、内存水位（stats 周期采样）、慢计数轨迹；扫描——连接数梯度（100/500/1000）× 速率梯度（10/100/1000 帧每秒）的 sweep——拐点（哪个维度先饱和：内存？扇出 CPU？Engine Worker？）指认扩容方向。容量规划产出——单实例的推荐负载带（如"≤800 连接且 ≤200 帧/秒时 P99 < 50ms"）+ 超出时的动作（加实例/降帧率/频道分片）。

复盘：项目五的新增经验

pushd 在十二条之上新增的经验：扇出设计先于优化（广播路径的引用发送/快照/跳过是设计决策不是性能补丁——架构期就要定）；长连接的终态对账（连接/组/订阅集三类状态的清理验证只能靠计数归零——第 110-112 章各层的"恰好一次"在项目层汇成一个对账表）；序号是廉价的自愈协议（一个单调整数换来漏检+re-sync——比确认重传机制便宜两个数量级，够用就是好协议）；测试客户端也是被测系统（并发测试框架的正确性不能靠时序假设——序号对账同样适用于测试代码自身）；运维接口是产品的一部分（列频道/踢出/慢计数不是"以后加"而是交付定义的一部分——第 104 章五段超时同理：可运维性在设计期决定）。五条累计十七条——最后一个项目（静态服务器收官）将检验它们在全栈组合下的韧性。

常见错误

n = xrtWsGroupCount(pGroup);
for ( i = 0; i < n; i++ ) { 发(第 i 个成员); }   /* 期间增删：错乱 */

xwsgroupsnapshot* Snap = xrtWsGroupSnapshotCreate(pGroup);
for ( i = 0; i < xrtWsGroupSnapshotCount(Snap); i++ ) {
	发送(xrtWsGroupSnapshotGet(Snap, i));
}
xrtWsGroupSnapshotDestroy(Snap);

for ( each member ) {
	while ( send(conn, msg) == AGAIN ) { wait(); }  /* 慢者拖全体 */
}

/* 组操作对象：全体受理即返回；慢者的 AGAIN 归其连接背压 */
xwsgroupop* Op = xrtWsGroupTextAsync(pGroup, msg);
xrtWsGroupOpDestroy(Op);   /* 发起即走（或按需等待聚合） */

stop_listener();
xrtWsGroupDestroy(pGroup);   /* 在途升级/连接未关：悬空与卡死 */

stop_listener();
xrtWsGroupSeal(pGroup);          /* 1.挡新 */
broadcast("shutting down");       /* 2.信号 */
close_all_members(pGroup);       /* 3.协议化关闭 */
wait_close_count_zero();         /* 4.终态计数 */
xrtWsGroupDestroy(pGroup);       /* 5.销毁 */
xrtNetEngineDestroy(pEngine);    /* 6.最后停 Engine */

编程练习

小结