从零实现跨进程共享内存无锁FIFO（ShmFifo）

简介：这份资源提供基于共享内存与信号量的C++版ShmFifo实现，聚焦进程间通信中的典型同步机制，适合正在学习操作系统、网络后台开发或嵌入式中间件的开发者，也适合作为IPC实践类课程设计参考。资源在C语言过程式shmfifo基础上，将一块共享内存与互斥量、full、empty三个信号量统一封装为ShmFifo类，并配套write、read、ipc与free等测试代码，能够帮助读者理解共享内存的创建、读写、释放流程，以及多信号量协同实现阻塞唤醒的并发控制思路。资源包共8个文件，含5个cpp源文件、2个头文件与1个Makefile，整体仅3KB，代码体量紧凑，便于快速通读、编译与二次修改，也适合与C语言版本对比学习面向对象封装带来的差异。目前已有286人浏览学习，对想要通过完整小项目掌握共享内存与信号量组合用法的读者来说，是一份直接可用的入门样例。 聊ShmFifo（Shared Memory FIFO）之前，先说一下我自己的背景。去年做一个图像采集项目，需要把多个采集进程产生的视频帧实时汇总到主控进程。一开始图省事直接用Unix Domain Socket，结果分辨率一上来，CPU就被拷贝和调度开销吃掉了大半。当时就意识到，单机多进程之间搬数据，如果还走内核，迟早要出事。后来把通信层换成基于共享内存的FIFO，带宽上去的同时CPU占用反而降下来了。这篇文章就是把当时封装ShmFifo（C++版）源码的过程完整复述一遍，从队列布局、无锁同步到各种坑，尽量把能说的细节都摊开讲。

如果你正在给某个C++项目做跨进程通信，或者单纯想了解共享内存队列应该怎么写，后面这部分内容应该对你有用。我会按源码设计的顺序走：为什么选共享内存、环形队列怎么设计、无锁同步怎么做、以及几个容易翻车的点。

1. 跨进程传数据，为什么最后选了共享内存FIFO

1.1 传统IPC方案的性能账

先算一笔账。管道、Unix Domain Socket、TCP本地回环，看起来都能传数据，但它们本质上是把数据从用户态拷贝到内核态，内核处理完后再拷贝到接收进程的用户态。也就是说，一次消息传递至少经历两次拷贝和一次调度切换，消息小的时候感觉不出来，消息频率一高，系统调用带来的上下文切换成本就很吓人。

我那个项目里每路视频大概是1080p30帧，一帧压缩后的数据量在几十KB到几百KB之间浮动，多路叠加每秒的数据量接近几百MB。用UDS去传这种流量，CPU一直都在忙着memcpy和陷入内核，真正干压缩、干编码的算力反而被挤占了。而且UDS还有背压问题，生产端写快了，接受端来不及消费，缓冲区一满，写操作要么阻塞要么返回错误，处理起来非常烦。

1.2 共享内存方案的本质优势

共享内存的思路很直接：操作系统把同一块物理内存映射到多个进程的地址空间，一个进程往里写，另一个进程直接读，中间不经过内核拷贝。数据从“生产者内存”到“消费者内存”的路径，从原来的两次内核拷贝变成了一次映射后的直接读写，跨进程通信的延迟因此低了一个甚至两个数量级。

再配合一个FIFO队列结构，生产者和消费者之间就形成了一个单向的、低延迟的数据通道。ShmFifo这个名字其实就是Shared Memory FIFO的缩写，它的核心价值在于：单机多进程之间，用尽可能少的CPU开销换尽可能高的吞吐。

我做过的对比测试大致是这样的（不同机器差异较大，但量级关系是稳定的）：

当然，共享内存不是银弹。它只能解决单机问题，不能跨机器；而且一旦进程崩溃，共享内存里的残留状态需要额外处理。但在“单机多进程高吞吐”这个具体场景下，它基本上是性能天花板。

2. 环形缓冲区与ShmFifo的内存布局

2.1 为什么是环形缓冲区

FIFO队列可以用链表、动态数组来实现，但在共享内存里，链表是一个糟糕的选择：节点动态分配意味着要自己写内存分配器，节点间指针在不同进程地址空间中映射地址可能不同，维护成本非常高。而环形缓冲区（Ring Buffer）在创建时一次性分配一整块连续内存，入队和出队都只搬数据，不需要任何动态分配，对缓存也友好。

环形缓冲区的设计思想很简单：把一整块线性内存首尾相连，通过两个游标标记读写位置。写入方往“尾部”写，读出方从“头部”读，两个游标往前移动，越过末尾就回到开头。于是整个队列的大小是固定的、预先分配的，并且不会出现“往中间插一个节点”这种操作，天然适合共享内存这种物理结构。

2.2 共享内存头部和元素区的真实布局

ShmFifo的共享内存映射区，我建议按“头部元数据 + 数据区”的格式来排布：

| ShmFifoHeader | 对齐填充 | T elements[capacity] |

头部里面记录的是所有进程协调工作所必需的元数据：读位置、写位置、容量、元素大小、数据区偏移、魔数。魔数的作用是判断这块共享内存是否已经被正确初始化过，避免后到的进程把一个未初始化的内存区域当成合法队列。

具体的头部结构，我用C++写了一个骨架版本，实际项目里可以在这个基础上扩展：

// ShmFifoHeader.h #pragma once #include <atomic> #include <cstdint> struct alignas(64) ShmFifoHeader { // 读游标：表示已经消费了多少个元素 // 单独占用一个cache line，避免与写游标互相干扰 std::atomic<uint64_t> read_pos{0}; char pad0[56]; // 写游标：表示已经生产了多少个元素 std::atomic<uint64_t> write_pos{0}; char pad1[56]; // 下面的字段在初始化后是只读的 uint64_t capacity; // 队列槽位数量 uint64_t elem_size; // 单个元素字节数 uint64_t payload_offset; // 数据区相对共享内存映射起始地址的偏移 uint32_t magic; // 魔数，用于判断是否已初始化 static constexpr uint32_t kMagic = 0x5348464F; // "SHFO" };

这里有一个非常关键的细节：read_pos和write_pos被alignas(64)和填充数据隔开，放在不同的缓存行里。因为生产者和消费者运行在不同的进程里，如果它们同时读写同一个缓存行上的不同字段，会造成缓存行乒乓（Cache Line Ping-Pong），性能会严重劣化。把两个游标隔开，是为了让它们各自占据独立的缓存行，减少不必要的缓存同步开销。

2.3 创建和初始化共享内存的核心流程

创建共享内存一般用POSIX接口shm_open+ftruncate+mmap。流程上有一个大家经常忽略的问题：两个进程可能同时执行创建逻辑，如果不加保护，可能导致二次初始化或者内存区域被重复截断。

我用的策略是：创建者负责初始化。具体实现是：

// 仅在创建者进程中执行的初始化函数 bool create_shared_memory(const char* name, size_t total_bytes, void** out_addr) { // O_EXCL 是关键：如果共享内存对象已存在，这里会失败 int fd = shm_open(name, O_CREAT | O_EXCL | O_RDWR, 0666); if (fd < 0) { // 已存在，走打开分支 fd = shm_open(name, O_RDWR, 0666); if (fd < 0) return false; } else { // 本进程是创建者，需要设置大小 if (ftruncate(fd, total_bytes) != 0) { close(fd); return false; } } void* addr = mmap(nullptr, total_bytes, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0); if (addr == MAP_FAILED) { close(fd); return false; } *out_addr = addr; close(fd); // mmap 之后 fd 可以关掉 return true; }

创建者初始化头部后，主动写入魔数；后到者发现魔数已经正确，就跳过初始化，直接进入使用阶段。这里有一点要注意：ftruncate设置的大小必须是你计算好的总字节数，而在调用mmap时，MAP_SHARED是必须的，否则多个进程之间的修改不会互相可见。

一个容易被忽略的问题：元素区大小要考虑对齐。payload_offset不能直接写成“sizeof(ShmFifoHeader)”就完事，因为如果T本身有严格的对齐要求（例如T是uint64_t或者某个alignas(16)的自定义结构体），数据区的起始地址必须是alignof(T)的整数倍。所以计算偏移时一般会做一步对齐处理：

size_t payload_offset = (sizeof(ShmFifoHeader) + alignof(T) - 1) / alignof(T) * alignof(T);

3. 无锁单生产者单消费者：同步机制的核心

3.1 用递增计数器代替环形游标

环形缓冲区的经典写法是在读写时对capacity取模。但ShmFifo这里可以更进一步：让read_pos和write_pos一直是单调递增的绝对计数器，只在计算内存偏移时才取模。

为什么可以这样设计？因为FIFO天然适合单生产者单消费者（SPSC）模型：写端只修改write_pos，读端只修改read_pos，两者从不写同一个变量，所以不需要锁。

具体判断逻辑是这样：

• 队列中可读数据量 =write_pos - read_pos
• 队列剩余空间 =capacity - (write_pos - read_pos)
• 队列空：write_pos == read_pos
• 队列满：write_pos - read_pos == capacity

有人会担心计数溢出。uint64_t的计数范围是2^64，就算每纳秒生产一个元素，也要跑几百年才可能溢出。实际工程里不用担心这个问题。

取模运算在计算偏移时会有优化空间：如果capacity是2的幂，模运算可以直接用位与操作替代，这也是后面性能优化中很重要的一环。

3.2 push/pop中的内存序设计与happens-before

无锁不等于没有顺序要求。为了保证生产者写入的数据在消费者那边一定可见，必须在游标更新时使用正确内存序。

这里用到的核心规则是C++11开始提供的std::atomic内存序：

• 生产者：先把元素写入数据区，然后以memory_order_release更新write_pos
• 消费者：先以memory_order_acquire读取write_pos，然后才读取数据区内容

release和acquire成对出现时，会形成happens-before关系：生产者在更新write_pos之前的所有写操作（也就是实际写入元素的动作），对成功读取到该write_pos新值的消费者来说是可见的。

push操作的核心逻辑如下：

bool push(const T& item) { // 写端读取自己的写游标用 relaxed 即可，因为没有竞争 uint64_t wp = header_->write_pos.load(std::memory_order_relaxed); // 读取读端的消费进度，需要 acquire，确保能看到消费方已释放的空间 uint64_t rp = header_->read_pos.load(std::memory_order_acquire); if (wp - rp >= header_->capacity) { return false; // 队列满 } T* slot = reinterpret_cast<T*>(payload() + (wp & mask_) * sizeof(T)); // 拷贝构造或直接赋值 *slot = item; // 写完数据后，release 更新写游标 header_->write_pos.store(wp + 1, std::memory_order_release); return true; }

pop操作与此对称：

bool pop(T& out) { // 读端读取自己的读游标也用 relaxed uint64_t rp = header_->read_pos.load(std::memory_order_relaxed); // 观察写端进度，需要 acquire，确保能看到生产者已发布的数据 uint64_t wp = header_->write_pos.load(std::memory_order_acquire); if (rp >= wp) { return false; // 队列空 } const T* slot = reinterpret_cast<const T*>(payload() + (rp & mask_) * sizeof(T)); out = *slot; // 数据读取完成后，release 更新读游标 header_->read_pos.store(rp + 1, std::memory_order_release); return true; }

很多第一次接触无锁队列的人会认为relaxed、acquire、release这些概念是性能玄学，其实不是。它们解决的是“编译器重排”和“CPU乱序执行”带来的可见性问题：如果没有这些内存序约束，编译器或CPU可能在实际运行中把写数据区和更新游标的顺序调换，消费者就可能读到“游标已经前进但数据还没写完”的脏状态。release/acquire的正确搭配，就能从根本上杜绝这个问题。

3.3 判空判满与边界条件

有人可能会问，wp - rp >= capacity这个判断够不够严谨？考虑一个极端情况：队列刚刚被填满，写端又快速调用了push。此时wp - rp == capacity，条件成立，返回false，表示满。没问题。

队列空的情况也一样：如果读端追上了写端，rp == wp，pop返回false。这里不会出现“读游标追上写游标导致误判”的经典环形缓冲陷阱，因为我们用的是绝对计数器而不是环形游标。绝对计数器天然规避了“满和空无法区分”的问题，这也是用递增计数器设计的一个巧妙之处。

4. 从源码到可用：五个最容易翻车的细节

4.1 不能往共享内存里塞std::string

这是我在实际项目中反复踩过的一个坑。std::string、std::vector这类容器内部有指针，指向堆上动态分配的内存。你把一个std::string对象写入共享内存，消费者看到的对象内存布局没问题，但里面的指针指向的是生产者进程的堆地址，消费者进程根本访问不了那块内存。

解决办法只有两个：一是所有跨进程传输的数据结构必须是自包含的（或者叫扁平化的），例如定长数组、std::array<char, N>，不要有任何堆指针；二是如果需要变长数据，采用“长度头 + 定长缓冲区”的序列化方案。定义模板时，最好加上类型约束：

static_assert(std::is_trivially_copyable_v<T>, "ShmFifo only supports trivially copyable types");

这个约束能挡住大部分不合适的类型。

4.2 跨进程使用std::atomic必须确认lock-free

C++标准规定std::atomic可以在多线程之间使用，但标准并没有明确保证它能被多个进程共享。在Linux上，如果std::atomic<T>是lock-free的，它本质上是直接操作内存地址，跨进程共享没有问题；但如果它内部带了一个锁（某些平台上更大的原子类型可能不是lock-free），那这个锁是进程私有的，另一个进程根本无法感知，整个同步就失效了。

所以我在代码里加了静态断言：

static_assert(std::atomic<uint64_t>::is_always_lock_free, "Need lock-free atomics for inter-process sharing");

注意用is_always_lock_free而不是is_lock_free，因为is_lock_free是运行时检查，而我们要的是编译期确定。在32位ARM上，uint64_t的原子操作可能不是lock-free，这时候要么换平台，要么退化成两个32位原子变量配合其他机制来设计，但复杂度会明显上升。

4.3 两个进程同时初始化怎么办

如果两个进程同时执行创建分支，都用O_CREAT | O_EXCL，只有一个会成功，另一个会失败然后转去打开已存在的对象。这个机制保证了初始化只有一个进程能做。但初始化本身也不是“一个函数调用”就结束的，它里面有多个写操作（写capacity、写payload_offset、写magic）。如果后到的进程在初始化还没完成的时候就读取magic，可能读到乱值。

我当时的处理很保守：先创建者初始化完所有字段，最后才写magic，并且用一下内存屏障确保顺序；后到者只有在确认magic正确后才继续使用。如果发现magic不对，就短暂重试。在实际项目中，这种竞争通常发生在进程启动阶段，只要init阶段有这层保护，后面就不会出问题。

4.4 进程崩溃后谁来清理

无锁SPSC的好处是：不需要锁，也就没有“持锁进程崩溃导致所有进程卡死”的问题。但这不代表没有残留状态。如果生产者写了一半崩溃，消费者可能会看到游标停在某个位置，数据区里有半截数据。这时候无法区分“真的没有数据”还是“崩溃残留”。

针对这个问题，一个实用的做法是在头部增加一个producer_pid字段和心跳时间戳。消费者发现长时间没有新数据写入时，可以检查生产者进程是否还存活，如果生产者已经退出，就主动重置队列游标或者通知上层做恢复处理。这个功能不是ShmFifo的核心，但在长跑的服务里非常有用。

4.5 权限、页对齐和shm_unlink时机

shm_open创建的共享内存对象默认权限受umask影响。如果你在代码里加了0666但系统umask是0022，其他用户实际上只有0644，没有写权限。跨用户调试时这个现象非常隐蔽，我是建议在测试时先把权限问题查清楚，别一上来就怀疑代码逻辑。

另外，mmap是按页映射的，共享内存对象的大小会被对齐到页大小。你ftruncate了1000字节，实际映射大小可能会大于1000字节，但你的代码逻辑仍然应该按自己计算的总字节数来访问，不要访问未初始化的区域。shm_unlink的时机也需要想清楚：它只是删除名字，不会立刻销毁实体内存，只有所有进程都munmap之后才会真正释放。所以我通常是在创建者退出时才调用shm_unlink，其他进程只munmap不删名。

5. 性能实测和还能榨出来的那点性能

5.1 一个简单的压测思路

写完一个ShmFifo之后，肯定要验证性能到底行不行。最简单的测试方式是准备两个进程：生产者进程不停往队列里写入带有序列号和时间戳的结构体，消费者进程读出来并记录时间差。把消息大小从64字节逐步增长到1MB，观察延迟和吞吐的变化。

在我自己的测试机上，64字节消息、队列容量设定为1024，单生产者单消费者情况下，单条消息的往返延迟可以稳定在亚微秒到几微秒范围内，吞吐量比UDS高出不少。但要提醒的是，绝对数值在不同机器上差别很大，关键看相对趋势：共享内存在大消息、高频率场景下的优势远比小消息场景更明显。

压测时有个容易忽略的点：生产者和消费者要分别绑定不同的CPU核心，否则操作系统调度器可能把两个进程调度到同一个核上反复切换，测出来的数据会非常难看。绑核用sched_setaffinity或者在启动时用taskset命令都可以。

5.2 走完这个项目后我认为值得做的优化

第一个优化是容量设为2的幂，把取模运算换成按位与。我上面代码里的mask_就是这么来的，这个改动对性能的提升虽然不算巨大，但在高频调用路径上很值得。

第二个优化是批量读写。把push改成push_batch(const T* items, size_t count)，一次更新一次游标，效果是显著的。尤其是传输小消息时，游标更新的开销占比很高，批量接口能把这部分成本摊薄到一个可接受的水平。

第三个优化是使用HugePage。当队列容量较大（几十MB以上），普通4KB页会导致TLB Miss频繁，这时可以用MAP_HUGETLB来分配大页内存。不过这个依赖系统配置，不是所有环境都能直接用，需要提前检查。

我在这个项目里最终的架构是：快路径使用ShmFifo传高频数据，慢路径保留UDS传控制消息和异常通知。两者分工明确，整体CPU占用比最初的全UDS方案降低了一半以上。

ShmFifo写起来不难，难的是把边界情况和跨进程语义想清楚。把它做好之后，你会发现单机多进程之间的数据传递其实可以非常轻快，轻快到几乎感觉不到通信开销的存在。如果你正在为跨进程数据传输犯愁，我建议你亲手实现一个最小的版本，跑一下压测，感受一下“数据在两进程之间直接流动”和“数据绕道内核再回来”的差异，那是一种很直观的体验。

本文还有配套的精品资源，点击获取