Unity MyFramework 用法说明（十二）：使用 DoubleBuffer 在多线程之间传递数据

上一篇介绍了SafeList它解决的是同一线程中遍历集合时增删元素的问题。但SafeList并不是线程安全集合。当子线程不断产生数据而主线程需要统一处理这些数据时MyFramework 提供了另一种工具DoubleBufferT项目地址https://github.com/ZHOURUIH/MyFramework一、DoubleBuffer 解决什么问题网络模块是最典型的使用场景。TCP 接收线程负责从 Socket 中读取并解析数据真正的协议逻辑则应该回到 Unity 主线程执行TCP 接收线程 ↓ 读取并解析网络数据 ↓ 将消息放入缓冲区 ↓ Unity 主线程 ↓ 取出消息并执行协议最直接的做法是使用一个加锁的Listprotected ListPacketReceiveInfo mReceiveList new(); protected ThreadLock mReceiveLock new(); public void addReceivePacket(PacketReceiveInfo packet) { using (new ThreadLockScope(mReceiveLock)) { mReceiveList.Add(packet); } }主线程读取时也必须加锁public void updateReceivePacket() { using (new ThreadLockScope(mReceiveLock)) { foreach (PacketReceiveInfo packet in mReceiveList) { executePacket(packet); } mReceiveList.Clear(); } }问题在于主线程遍历和执行消息期间接收线程无法继续写入。如果协议数量较多或者消息处理逻辑较复杂锁会被长时间占用。DoubleBuffer的目的就是只在交换缓冲区时短暂加锁读取和遍历过程中允许其他线程继续写入。二、DoubleBuffer 的工作方式DoubleBuffer内部维护两个列表列表 A 列表 B任意时刻一个负责写入另一个负责读取写线程 → 列表 A 主线程 → 列表 B主线程准备读取时框架会交换两个列表交换前 写列表A 读列表B 交换后 写列表B 读列表A交换操作在锁中快速完成。交换完成以后主线程遍历列表 A 子线程继续向列表 B 写入两个线程不再操作同一个列表所以主线程遍历期间不需要一直持有锁。下一次读取时再交换回来写列表A 读列表B这就是双缓冲的核心。三、创建 DoubleBuffer以网络接收消息为例protected DoubleBufferPacketReceiveInfo mReceiveBuffer new();PacketReceiveInfo用来保存一次网络消息的基本信息public struct PacketReceiveInfo { public byte[] mPacketData; public ulong mFieldFlag; public int mPacketSize; public uint mSequence; public ushort mType; public bool mHasSign; public PacketReceiveInfo( byte[] data, ulong fieldFlag, int packetSize, uint sequence, ushort type, bool hasSign) { mPacketData data; mFieldFlag fieldFlag; mPacketSize packetSize; mSequence sequence; mType type; mHasSign hasSign; } }这里使用结构体保存消息的中转信息避免再创建一个额外的普通对象。四、在子线程中写入数据接收线程解析出一个完整的数据包后直接调用mReceiveBuffer.add(new PacketReceiveInfo( packetData, fieldFlag, packetSize, sequence, packetType, hasSign));add()可以从多个线程调用。它会短暂锁住当前写列表把数据加入以后立即释放锁线程获得锁 ↓ 向当前写列表添加数据 ↓ 释放锁写线程不需要关心当前哪个列表是读列表也不需要判断主线程是否正在遍历。五、在主线程中读取数据主线程使用DoubleBufferReaderpublic void update(float elapsedTime) { using var reader new DoubleBufferReaderPacketReceiveInfo( mReceiveBuffer); foreach (PacketReceiveInfo info in reader.mReadList) { executePacket(info); } }创建DoubleBufferReader时会调用mReceiveBuffer.get();get()会交换读写列表并返回刚刚停止写入的列表。离开作用域后DoubleBufferReader.Dispose()会自动清空本次读取的列表 结束读取状态因此不需要手动调用reader.mReadList.Clear(); mReceiveBuffer.endGet();六、完整的简化示例下面是一个简化的网络消息中转类public class NetworkReceiveQueue : FrameSystem { protected DoubleBufferPacketReceiveInfo mReceiveBuffer new(); // 可以从网络接收线程调用 public void pushPacket(PacketReceiveInfo packet) { mReceiveBuffer.add(packet); } // 由主线程统一执行 public override void update(float elapsedTime) { base.update(elapsedTime); using var reader new DoubleBufferReaderPacketReceiveInfo( mReceiveBuffer); foreach (PacketReceiveInfo packet in reader.mReadList) { executePacket(packet); } } public override void destroy() { mReceiveBuffer.destroy(); base.destroy(); } protected void executePacket(PacketReceiveInfo packet) { // 在主线程解析并执行协议 } }这样网络线程只负责生产数据mNetworkReceiveQueue.pushPacket(packet);Unity 主线程只负责消费数据读取这一批消息 执行这一批消息 清空读取列表 等待下一帧两边的职责非常明确。七、为什么不直接交换一个普通 List也可以自己维护两个列表protected ListPacketReceiveInfo mListA new(); protected ListPacketReceiveInfo mListB new();但还需要处理当前写入的是哪一个列表 当前读取的是哪一个列表 什么时候允许交换 交换过程如何加锁 读取期间能否再次交换 是否在同一个线程读取 读取完成后如何清空 销毁时如何处理DoubleBuffer把这些规则统一封装起来业务代码只需要mReceiveBuffer.add(packet);以及using var reader new DoubleBufferReaderPacketReceiveInfo( mReceiveBuffer);八、手动读取方式不使用DoubleBufferReader时也可以手动操作ListPacketReceiveInfo readList mReceiveBuffer.get(); foreach (PacketReceiveInfo packet in readList) { executePacket(packet); } readList.Clear(); mReceiveBuffer.endGet();但这种写法存在一个问题ListPacketReceiveInfo readList mReceiveBuffer.get(); if (readList.Count 0) { return; }提前返回后endGet()没有执行缓冲区会一直处于读取状态。所以下一次调用get()时框架会提示读列表正在使用中不能再次获取读列表因此更推荐使用using var reader new DoubleBufferReaderPacketReceiveInfo( mReceiveBuffer);即使中途return或出现异常也能正常结束读取。九、同一个 DoubleBuffer 只能有一个读取线程DoubleBuffer支持多个线程写入 一个固定线程读取不支持多个线程同时读取 同一个线程重复获取读列表 读取过程中再次调用 get()第一次调用get()时框架会记录读取线程的 ID。以后必须继续由这个线程读取网络线程1 ─┐ 网络线程2 ─┼→ add() 网络线程3 ─┘ Unity主线程 → get()不能这一次在主线程读取下一次又切换到其他工作线程读取。同一个DoubleBuffer也不能嵌套创建两个 Readerusing var readerA new DoubleBufferReaderPacketReceiveInfo( mReceiveBuffer); // 错误第一个 Reader 还没有结束 using var readerB new DoubleBufferReaderPacketReceiveInfo( mReceiveBuffer);读取列表只能在当前 Reader 的作用域中使用。十、数据本身的生命周期仍然要管理DoubleBuffer只保证列表读写安全不会自动管理元素内部的数据。例如网络消息中包含一个线程对象池分配的数组PacketReceiveInfo info new( packetData, fieldFlag, packetSize, sequence, packetType, hasSign); mReceiveBuffer.add(info);将数据加入缓冲区以后写线程不能立即回收mReceiveBuffer.add(info); // 错误主线程还没有处理这个数组 UN_ARRAY_BYTE_THREAD(info.mPacketData);应该等主线程处理完成后再回收foreach (PacketReceiveInfo info in reader.mReadList) { NetPacket packet parsePacket( info.mType, info.mPacketData, info.mPacketSize, info.mSequence, info.mFieldFlag, info.mHasSign); UN_ARRAY_BYTE_THREAD(info.mPacketData); packet?.execute(); }向缓冲区写入数据实际上也代表把这份数据的处理权交给了读取线程。十一、限制写缓冲区大小可以设置写列表的最大数量mReceiveBuffer.setWriteListLimit(1024);当当前写列表已经达到上限时后续数据不会继续写入。这可以避免生产速度远高于消费速度时缓冲区无限增长。但需要注意add() 没有返回是否写入成功 超过限制的数据会被忽略所以不能丢失的数据不应该随意设置一个过小的上限。如果业务必须限制数量还需要结合自己的逻辑记录溢出次数或处理异常状态。默认值为00 表示没有数量上限十二、什么时候调用 clear 和 destroy清空缓冲区mReceiveBuffer.clear();它会同时清空读列表和写列表。但不能在 Reader 仍然存在时调用using var reader new DoubleBufferReaderPacketReceiveInfo( mReceiveBuffer); // 错误当前仍在使用读列表 mReceiveBuffer.clear();销毁时调用mReceiveBuffer.destroy();销毁前应该先停止所有可能继续调用add()的工作线程停止生产线程 ↓ 确认不再写入 ↓ 处理或释放缓冲区中的数据 ↓ 销毁 DoubleBuffer不能在子线程仍然运行时直接销毁缓冲区。十三、DoubleBuffer 和 SafeList 的区别两者解决的问题不同。SafeList用于同一个线程内正在遍历集合 ↓ 调用链中可能增加或删除元素 ↓ 当前遍历保持稳定典型场景子弹更新 怪物更新 状态更新 可见格子更新DoubleBuffer用于多个线程之间一个或多个线程不断生产数据 ↓ 一个固定线程批量读取数据典型场景网络线程 → 主线程 文件读取线程 → 主线程 日志线程 → 写入线程 后台计算线程 → 结果处理线程可以简单理解为SafeList 解决遍历过程中修改集合 DoubleBuffer 解决生产线程与消费线程之间的数据交换十四、总结MyFramework 中DoubleBuffer的基本用法是protected DoubleBufferMyData mBuffer new();生产线程写入mBuffer.add(data);消费线程读取using var reader new DoubleBufferReaderMyData(mBuffer); foreach (MyData data in reader.mReadList) { process(data); }它通过两个列表交替承担读写任务写线程持续写入一个列表 ↓ 消费线程交换两个列表 ↓ 消费线程无锁遍历旧写列表 ↓ 生产线程继续写入新列表DoubleBuffer并不是完全没有锁而是把锁限制在数据写入和缓冲区交换的短暂过程里避免消费线程在处理整批数据时长期占用同一把锁。对于“多个线程生产一个线程批量消费”的场景这种结构既能保证数据交换安全也能减少读写双方互相等待的时间。