第 21 章 无锁队列：SPSC / MPSC / MPMC

本章导读

队列是卷三唯一线程安全的容器——它本来就是为跨线程传数据而生的，也因此是卷三与卷六（并发）之间的桥：本章教容器的用法与纪律，卷六把它接进线程、协程与 Channel 的完整体系——先在这里把单件武器用熟，到卷六才谈阵型。XRT 提供三种拓扑：SPSC（单生产者单消费者，最快）、MPSC（多生产者单消费者，事件聚合的标准形）、MPMC（任意多对多，最通用也最贵）。三者共享一套使用语言：绑定外部槽位（零内部分配）、批量收发（PushBatch/PopBatch）、关闭停机协议（Close 后生产者停、消费者取尽）。本章把这三件事讲透，并给出"拓扑优先"的选型纪律——通用性是有价格的，能确定拓扑就不要为它付钱。

21.1引入

一个日志线程要接收全部工作线程的日志事件——这是 MPSC：多个生产者、一个消费者。一个渲染线程与一个解码线程一对一传帧——这是 SPSC。一个任务分发器让任意线程投递、任意线程领取——这是 MPMC。三个场景如果都用"一把互斥锁包一条普通队列"解决，能跑，但每秒百万次收发的路径上，锁的争用会把多核并行的优势磨平——十核机器八个核在等锁是常见惨案；而锁方案与无锁方案的差异，在本章练习的对照测量里会给出你机器上的具体数字。

无锁队列用原子操作替代锁：生产者与消费者各自推进自己的指针、以原子发布让对方可见——SPSC 甚至只需要两个原子的头尾指针，单条中位数百万级每秒；MPMC 要为每个槽位做发布同步，单条最贵、但批量接口把摊还成本压下来（仓库基准：批量 32 条时 1.34 亿条每秒）。拓扑越确定，同步越少——这就是"拓扑优先"：日志聚合确定是 MPSC 就用 MPSC，不要"以防万一"上 MPMC。

21.2概念

三种拓扑一张表

三个共享语言

绑定槽位意味着队列的存储由调用方提供（栈数组、静态区、池都行）——队列只提供同步与索引逻辑，这也是 _noalloc 契约测试覆盖的形态之一。批量接口的返回值是"实际完成数"（xqueuebatchresult 的 Count）——队列可能未满但你请求的部分成功，返回值告诉你推进到了哪里。关闭停机是并发程序的终止协议核心：Close 必须在"确认全部生产者结束"后调用；已入队的数据不丢，消费者继续取尽。

FIFO 与饥饿

三条队列都是 FIFO：先入先出、跨批保序。MPMC 的多个消费者并发 PopBatch 时，队列内部保证每条消息恰好被领取一次——不会重复投递，也不需要消费者协调。"公平性"则不做承诺：高频消费者可能长期领得多——需要加权公平就在业务层做（配额计数），不指望队列。

与卷六的关系

本章只教"单机多线程之间传指针"——跨进程的传输属于卷七网络与序列化的领地。卷六会把队列放进完整的并发体系：与协程调度器的配合（阻塞等待变成可等待的 Pop）、Channel 的"带关闭语义的队列"封装、以及什么时候该用 Future 代替队列。现在只需要记住：队列传的是 ptr——消息对象的生命周期归发送方分配、接收方处理，跨过队列边界后不要再由发送方释放。

21.3示例

完整程序：MPMC 批量收发与停机

来自仓库范例 examples/containers/queue_mpmc/main.c：

 */

#include <stdio.h>
#include <xrt.h>



int main(void)
{
	xmpmcqueue Queue;
	xqueueslot Storage[8];
	int pValues[] = { 10, 20, 30, 40 };
	ptr pFirstBatch[] = { &pValues[0], &pValues[1] };
	ptr pSecondBatch[] = { &pValues[2], &pValues[3] };
	ptr pOutput[4];
	xqueuebatchresult Batch;

	/* 绑定外部槽位；MPMC 槽位与 MPSC 同类型（xqueueslot）。 */
	if ( !xrtMPMCQueueInitBuffer(&Queue, Storage, 8u) ) {
		return 1;
	}

	/* 两批各 2 条：并发场景下任意线程都可这样批量发布。 */
	if ( xrtMPMCQueuePushBatch(&Queue, pFirstBatch, 2u).Count != 2u ) {
		xrtMPMCQueueUnit(&Queue);
		return 2;
	}
	if ( xrtMPMCQueuePushBatch(&Queue, pSecondBatch, 2u).Count != 2u ) {
		xrtMPMCQueueUnit(&Queue);
		return 3;
	}

	


	 */
	xrtMPMCQueueClose(&Queue);

	


	 */
	Batch = xrtMPMCQueuePopBatch(&Queue, pOutput, 4u);
	for ( size_t i = 0; i < Batch.Count; i++ ) {
		printf("%d\n", *(int*)pOutput[i]);
	}
	xrtMPMCQueueUnit(&Queue);
	return Batch.Count == 4u ? 0 : 4;
}

刚才发生了什么。 ① xqueueslot Storage[8] 是调用方提供的槽位数组，InitBuffer(&Queue, Storage, 8) 绑定——容量 8 由数组长度决定，队列自身零分配。② 两批各 2 条 PushBatch：返回值 .Count 必须核对——"请求 2 条实际进了几条"在满载时是真实问题，程序按实际数推进。③ Close 在全部生产者结束后调用——单线程示例里它紧随其后，多线程程序里它属于停机协议（等待生产者线程 join 后再 Close）。④ PopBatch 一次领 4 条，输出严格 FIFO：10、20、30、40——跨批的顺序保持。多消费者并发执行同一个 PopBatch 时内部保证每条恰好一次。⑤ 消息是 ptr：队列里的 &pValues[i] 指向栈数组——接收方只借用不释放，对象归属纪律与第 5 章一致。

完整程序：SPSC 最小形态

来自 examples/containers/queue_spsc/main.c，一对一拓扑的最快路径：

 */

#include <stdio.h>
#include <xrt.h>



int main(void)
{
	xspscqueue Queue;
	ptr Storage[8];          /* 槽位由调用方提供：地址稳定、零分配 */
	int pValues[] = { 10, 20, 30 };
	ptr pValue;

	/* 绑定外部槽位：容量 8，Push/Pop 全在这块内存上进行。 */
	if ( !xrtSPSCQueueInitBuffer(&Queue, Storage, 8u) ) {
		return 1;
	}

	/* 依次入队三个 int 指针（队列搬运的是 ptr，不拥有对象）。 */
	for ( size_t i = 0; i < 3u; i++ ) {
		if ( xrtSPSCQueueTryPush(&Queue, &pValues[i]) != XQUEUE_OK ) {
			xrtSPSCQueueUnit(&Queue);
			return 2;
		}
	}

	



	 */
	xrtSPSCQueueClose(&Queue);
	while ( xrtSPSCQueueTryPop(&Queue, &pValue) == XQUEUE_OK ) {
		printf("%d\n", *(int*)pValue);
	}
	xrtSPSCQueueUnit(&Queue);
	return 0;
}

刚才发生了什么。 接口形态与 MPMC 完全同构——绑定槽位、批量收发、关闭停机三件套一个不少——差异全在内部：SPSC 只有一头一尾两个原子指针，生产者写头、消费者写尾，互不争用。同样的消息量它单条开销最小，这就是"拓扑确定时选专用队列"的兑现。把两份示例的接口调用并排读一遍：学一种用法，三种队列通用——这是队列家族 API 设计的用心之处。

容量与背压的第一次相遇

队列容量（槽数组长度）不是越大越好，它是背压的第一道阀门：队列满时 PushBatch 部分成功，生产端感知到推不动就该减速或缓冲——这正是第 68 章网络背压的微缩版。容量规划的两个极端都有代价：容量太小，突发流量瞬间打满，生产者频繁空转重试；容量太大，内存驻留上涨，而且消费者故障时积压的消息更多、恢复更久。经验起点：容量取消费者一个调度周期处理量的 2 到 4 倍，再用第 6 章统计与队列水位实测校准。队列的 Count 是公开字段——生产端的快满检查零成本。

无锁的纪律边界

无锁指内部不用互斥锁，不等于随便并发——两条纪律仍然成立。其一，SPSC 的两端各自独占：一个生产者线程、一个消费者线程；多线程 Push 同一条 SPSC 是未定义行为，那是 MPSC 的场景。其二，槽内数据的发布时序由队列保证，但你自己的消息对象必须入队前写好、出队后才写——跨过队列边界的写竞争不是队列的责任。把这两条纪律记牢，无锁队列的快才真正属于你。

21.4契约

• 绑定存储：槽位数组由调用方提供；容量 = 槽数组长度；队列零内部分配。
• 批量核对：Push/Pop 的返回 Count 是实际完成数，满载与空载都可能部分成功。
• 停机协议：Close 在全部生产者结束后调用一次；已入队数据可继续取尽；关闭后 Push 失败。
• 恰好一次：多消费者并发 Pop，每条消息恰好被领取一次，无需外部去重。
• 消息归属：队列传 ptr；对象由发送方创建、接收方处理，归属纪律同第 5 章；入队前写好、出队后才写。
• 拓扑优先：能确定一对一/多对一就选 SPSC/MPSC；MPMC 为通用性付最高单价。

选型速断：三问定队列

实际选型时三个问题按序问自己。第一问：几条流水线？ 两条线程一对一，SPSC 到手，结束。第二问：谁聚合？ 多个源汇到一个处理者（日志、监控、事件循环），MPSC。第三问：真的任意对任意吗？ 大量写着 MPMC 的场景，细看其实是"多个生产者 + 若干消费者池"——消费者数量固定且任务等价时，MPSC 到分发器、分发器 SPSC 到各消费者，两级专用队列常常比一级 MPMC 更快也更清晰。三问答完还剩 MPMC 的，才是真需要它的场景：动态工作窃取、无中心分发的任务图。把这三个问答写进设计文档的选型段落，评审时就有据可查。

常见错误

xrtMPMCQueuePushBatch(&Queue, pMessages, 32u);   /* 返回值被丢弃 */
/* 队列可能只收下前 N 条——剩下 32-N 条静默滞留在发送缓冲 */

xqueuebatchresult Sent = xrtMPMCQueuePushBatch(&Queue, pMessages, 32u);
if ( Sent.Count < 32u ) {
	RetryLater(pMessages + Sent.Count, 32u - Sent.Count);   /* 从断点续传 */
}

/* 主线程：不等工作线程 */
xrtMPMCQueueClose(&Queue);
/* 某工作线程此刻还在 Push——停机协议被破坏 */

for ( i = 0; i < WorkerCount; ++i ) {
	xrtThreadWait(Workers[i], NULL);   /* 先等全部生产者结束（第 56 章线程 API） */
}
xrtMPMCQueueClose(&Queue);            /* 再关闭 */
DrainAndShutdown(&Queue);             /* 消费者取尽后自然终止 */

编程练习

小结