用途:该技能说明如何为 bocpy 消息队列编写测试。该队列是无锁、基于标签的 MPSC 环形缓冲区,由 C 实现,是 send()/receive() 的底层。
架构与主要能力:共 16 个静态分配的环形缓冲区(BOC_QUEUE_COUNT = 16),每个容量 16384。标签为字符串,与队列一一对应,每个唯一标签占用一个队列槽位。分配模式分自动(默认,首次 send/receive 时标签占用队列)与显式(set_tags() 预分配标签、清除所有消息并重置队列)。set_tags([]) 会清空所有队列、释放标签关联并将队列全部置为 UNASSIGNED,恢复自动分配。入队无锁(对 tail 做原子 CAS),多线程可并发向同一标签发送;receive 在队列上自旋并以睡眠等待消息或超时。
使用方式:测试文件位于 test/test_message_queue.py。每个测试需清理状态,使用 autouse fixture 在测试前后调用 set_tags([]):
MAX_QUEUES = 16
@pytest.fixture(autouse=True) def reset_queues(): set_tags([]) yield set_tags([])
fixture 目的是清除陈旧状态而非预分配标签,多数测试可自然走自动分配路径。
典型测试模式:基础发送/接收并断言;验证同一标签的 FIFO 顺序;跨多标签的选择性接收(receive 接受标签列表/元组,返回首个可用匹配,非匹配标签的消息保留);超时行为(超时值 0 立即返回或 (TIMEOUT, None),正浮点最多等待相应秒数,负数默认无限阻塞;可选 after 回调仅在超时时触发并替换默认返回值);跨线程消息传递;多并发生产者(每生产者 FIFO 有序);标签容量上限(超过 16 个抛 KeyError);显式 set_tags 行为;负载往返保真(通过跨解释器数据 XIData 或 pickle)。
错误处理:非字符串标签发送抛 TypeError;receive 传入空标签列表抛 RuntimeError;标签列表含非字符串抛 TypeError;超过唯一标签容量抛 KeyError;set_tags 标签过多抛 IndexError;set_tags 含非字符串或非序列抛 TypeError。
常见陷阱:调用 receive 不传超时会导致测试挂起,应始终传超时;跨测试使用过多唯一标签会耗尽队列槽,需用 fixture 重置;忘记 fixture 会残留消息或标签分配导致不稳定失败;set_tags 并非必需,仅在需预分配或清空时调用;断言前应 join 线程;并发线程若使用不同标签会触及 16 队列上限,应共享同一标签。
限制:资料未提供安装命令、收费、价格或平台信息,暂未提供。