简介ShmFifo源代码C版是一份聚焦共享内存与信号量协同工作的进程间通信示例适合正在学习Linux IPC机制、需要实现多进程高效数据交换的开发者。资源由C语言过程调用版本改造而来把一块共享内存与互斥、满、空三个信号量封装成C类解决读写同步与资源竞争问题提供从初始化、写入、读取到释放的完整闭环。包内共8个文件含5个cpp源码、2个头文件和1个makefile构建脚本cpp文件覆盖核心类实现与读写释放操作头文件声明接口makefile支持一键构建压缩包仅3KB结构紧凑、无冗余依赖便于直接阅读和交叉编译。已有288人浏览学习。研读后可掌握多个信号量协同管理共享内存的设计思路理解C用构造析构封装底层IPC细节的方式并能基于自带读写测试代码快速验证队列行为是进阶学习生产者消费者模型和共享内存应用的实用参考。1. 共享内存 FIFO先搞清楚它解决了什么问题做后台服务的人早晚会遇到这么个需求两个独立进程之间要高频传消息又不能上关系型数据库也不能为了传个几 KB 的数据就开一个 TCP 连接。我第一次是在一个采集模块和存储模块之间面临这个选择——采集进程不停写存储进程稍慢一点读中间要缓冲还不能丢数据。搜了一圈源码方案最后发现共享内存 信号量做的 FIFO 队列最合适。ShmFifo 就是这类实现的 C 封装一块共享内存当环形缓冲区两个信号量控制读和写的节奏进程 A 往里塞数据进程 B 按序取出来延迟在微秒级完全绕开内核协议栈。适合做高频数据管道、日志转发、进程间命令队列或者任何一对多、多对一的实时数据中转。这篇文章把设计选型、核心代码、以及导致翻车的细节一次讲清楚。2. 设计原理与选型环形缓冲 信号量如何做到进程安全2.1 为什么不选管道和消息队列两种 IPC 的边界条件管道pipe是很多人第一个想到的方案。它的好处是简单fork之后直接write/read就能通信。但管道有个硬伤它的缓冲区在内核空间容量默认只有 64 KB 且不可配置Linux 上通过fcntl改F_SETPIPE_SZ也只是到 1 MB 量级数据频繁跨内核边界拷贝。两个进程跑在同一台机器上这一趟拷贝完全没必要。更麻烦的是管道是字节流没有消息边界你要自己设计分隔符或长度头协议解析多一层负担。消息队列Message Queue用msgget/msgsnd/msgrcv自带消息边界和优先级看似很合适。但 System V 消息队列每条消息有 8 KB 的上限MSGMNB是队列总字节数MSGMNI是系统级队列数高吞吐下容易撞限额。而且消息队列的接口太陈旧msgrcv的拷贝同样是先到内核再回用户态。真正的高频场景里这两个方案在性能上都不如用户态共享内存。共享内存的思路是把一块物理内存映射到两个进程的虚拟地址空间写入方直接memcpy读取方直接memcpy全程没有内核参与数据移动。剩下的问题只有一个怎么保证两个进程不会同时踩同一块区域、不会读到半个消息。这就要靠信号量来做同步。2.2 环形缓冲区与内存布局容量怎么定、指针怎么转共享内存 FIFO 的存储区我习惯做成环形缓冲区避免频繁搬移数据。先看内存布局字段类型说明uint32_t magic4 字节校验是否为有效 ShmFifo 区域uint32_t capacity4 字节环形数据区容量单位字节uint32_t write_pos4 字节写偏移只由生产者修改uint32_t read_pos4 字节读偏移只由消费者修改uint8_t data[]可变实际数据区长度等于 capacity写偏移和读偏移不直接用指针而用相对数据区起始的字节偏移。原因是两个进程映射到共享内存的基地址不一定相同一个是0x7f1234568000另一个可能是0x7fabcdef0000绝对指针没法直接用偏移量却是共同的。每次定位数据时这样换算uint8_t* base static_castuint8_t*(shm_ptr); uint8_t* data_area base sizeof(ShmFifoHeader); uint8_t* write_addr data_area write_pos;数据结构本身用sem_t实现同步。这里提一个设计取舍我用了三个信号量而不是两个。sem_empty初值等于 capacity表示空闲字节数sem_full初值为 0表示已写入的字节数sem_mutex初值为 1保护write_pos和read_pos的更新。为什么要第三个sem_mutex因为write_pos和read_pos会回绕特殊的满/空状态write_pos read_pos区分不出来而且多进程同时写时两个memcpy交错执行可能破坏数据。sem_empty和sem_full只解决整体节奏sem_mutex解决同一时刻只有一个进程改指针。这是最稳的组合也最容易向团队成员解释。容量capacity的设定原则必须大于单条消息最大长度的两倍。如果单条 32 KB容量至少 64 KB否则写入方会在空间不足时阻塞读者来不及消费就死锁。我一般再往对齐到 4096 的整数倍因为共享内存页大小就是 4 KB容量不对齐会浪费一页或者跨页降低映射效率。3. 源码拆解从一个可编译的 ShmFifo 类说起3.1 核心结构体与初始化流程这段代码是 Linux 下的 POSIX 共享内存方案用shm_open创建共享内存对象再mmap到进程地址空间。相比 System V 的shmgetPOSIX 方式的接口更干净不需要维护ipcs遗留的 key也方便用文件描述符做生命周期管理。#include fcntl.h #include sys/mman.h #include semaphore.h #include unistd.h #include cstring #include stdexcept #include string struct ShmFifoHeader { uint32_t magic; uint32_t capacity; uint32_t write_pos; uint32_t read_pos; }; class ShmFifo { public: // create: true 表示创建并初始化false 表示仅连接已存在的实例 bool init(const char* shm_name, uint32_t capacity, bool create) { int oflag create ? (O_CREAT | O_RDWR) : O_RDWR; int fd ::shm_open(shm_name, oflag, 0644); if (fd 0) return false; if (create) { if (::ftruncate(fd, sizeof(ShmFifoHeader) capacity) ! 0) { ::close(fd); return false; } } void* mapped ::mmap(nullptr, sizeof(ShmFifoHeader) capacity, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0); ::close(fd); // mmap 之后 fd 可以关闭 if (mapped MAP_FAILED) return false; header_ static_castShmFifoHeader*(mapped); data_ static_castuint8_t*(mapped) sizeof(ShmFifoHeader); if (create) { header_-magic 0x5348464F; // SHFO header_-capacity capacity; header_-write_pos 0; header_-read_pos 0; // 信号量也要创建名字与共享内存关联 sem_open_full_name_ std::string(shm_name) _full; sem_open_empty_name_ std::string(shm_name) _empty; sem_open_mutex_name_ std::string(shm_name) _mtx; sem_unlink(sem_open_full_name_.c_str()); sem_unlink(sem_open_empty_name_.c_str()); sem_unlink(sem_open_mutex_name_.c_str()); sem_full_ sem_open(sem_open_full_name_.c_str(), O_CREAT | O_EXCL, 0644, 0); sem_empty_ sem_open(sem_open_empty_name_.c_str(), O_CREAT | O_EXCL, 0644, capacity); sem_mutex_ sem_open(sem_open_mutex_name_.c_str(), O_CREAT | O_EXCL, 0644, 1); } else { // 连接方只打开不创建避免重复初始化计数器 sem_full_ sem_open(sem_open_full_name_.c_str(), 0); sem_empty_ sem_open(sem_open_empty_name_.c_str(), 0); sem_mutex_ sem_open(sem_open_mutex_name_.c_str(), 0); } if (!sem_full_ || !sem_empty_ || !sem_mutex_) return false; return true; } private: ShmFifoHeader* header_ nullptr; uint8_t* data_ nullptr; sem_t* sem_full_ nullptr; sem_t* sem_empty_ nullptr; sem_t* sem_mutex_ nullptr; std::string sem_open_full_name_; std::string sem_open_empty_name_; std::string sem_open_mutex_name_; };初始化这段有几个关键动作值得说透。ftruncate必须在mmap之前执行它决定了共享内存对象的实际大小mmap映射的长度不能超过文件大小否则访问时直接 SIGBUS。我先对信号量做一次sem_unlink再sem_open这是为了清掉上次进程异常退出残留的信号量状态。如果不清理新进程拿到的可能是一个初值不对的信号量比如sem_full残留为 5新进程以为缓冲区里有 5 字节数据实际上共享内存早就被销毁重建了。连接方createfalse的信号量直接用sem_open(name, 0)打开不带O_CREAT。这里的前提是创建方必须在连接方之前启动。如果顺序反过来连接方会打开失败。我在实际项目里会让消费者进程等待生产者进程先创建通过一个while (shm_open ENOENT) usleep(1000)的循环来规避启动时序问题。3.2 写入与读取两条路径的代码实现写入路径的逻辑分四步申请空位、加互斥锁、拷数据、释放锁并通知读者。这里有一个容易写错的地方memcpy必须在获取sem_mutex之后执行不能在只拿到sem_empty后就开拷。sem_empty保证的数据总量足够但不保证别的写者当前没在写两个写者可能同时在同一个位置附近操作。bool ShmFifo::write(const void* data, uint32_t len) { if (!data || len 0 || len header_-capacity) { return false; // 单条超过容量直接失败不阻塞 } // 第一步等空位 if (::sem_wait(sem_empty_) ! 0) return false; // 第二步进入临界区更新写指针 if (::sem_wait(sem_mutex_) ! 0) { ::sem_post(sem_empty_); // 拿不到锁要归还空位 return false; } uint32_t wpos header_-write_pos; uint32_t cap header_-capacity; uint32_t tail cap - wpos; // 从写位置到缓冲区末尾的剩余空间 if (len tail) { std::memcpy(data_ wpos, data, len); } else { // 数据跨过缓冲区尾部需要拆成两段写 std::memcpy(data_ wpos, data, tail); std::memcpy(data_, static_castconst uint8_t*(data) tail, len - tail); } // 回绕新位置一定小于 capacity因为 len capacity header_-write_pos (wpos len) % cap; // 第三步释放锁通知读者有数据 ::sem_post(sem_mutex_); ::sem_post(sem_full_); return true; }这段代码的边界条件是重头戏。tail cap - wpos是当前线性剩余空间当len tail时第一段拷贝到缓冲区尾部第二段从头开始总量tail (len - tail)正好等于len。回绕用取模而不是if判断因为(wpos len)可能等于cap取模后变为 0正好落回起点。单条len超过容量的情况我选择直接返回失败而不是等消费者腾空间原因后面避坑章细说。读取路径是写入的镜像操作顺序反过来先等sem_full再拿sem_mutex最后拷贝。bool ShmFifo::read(void* buf, uint32_t len) { if (len header_-capacity) return false; if (::sem_wait(sem_full_) ! 0) return false; if (::sem_wait(sem_mutex_) ! 0) { ::sem_post(sem_full_); // 归还数据信号量 return false; } uint32_t rpos header_-read_pos; uint32_t cap header_-capacity; uint32_t tail cap - rpos; if (len tail) { std::memcpy(buf, data_ rpos, len); } else { std::memcpy(buf, data_ rpos, tail); std::memcpy(static_castuint8_t*(buf) tail, data_, len - tail); } header_-read_pos (rpos len) % cap; ::sem_post(sem_mutex_); ::sem_post(sem_empty_); // 读走 len 字节释放 len 字节空位 return true; }实现read有个隐含假设调用方提前知道自己要读多少字节。这在定长消息场景下没问题比如传感器数据固定 24 字节或日志一条 4 KB。但如果消息变长读取端不知道长度常见做法是写两段先写 4 字节长度头再写数据体。读取方先读 4 字节得到len再读len字节数据体。这个约定要在通信协议里固定下来不属于 ShmFifo 本身的功能但设计时必须留出len的写入空间否则容量规划会算漏。两个read连续调用是有问题的第一次读了 4 字节长度头第二次读数据体这中间缓冲区可能已经有新数据被别的消费者读走消息就错位了。所以变长消息的读写必须放在同一个互斥区间内。我在这套实现里没有给read和write提供批量锁因为sem_empty和sem_full的粒度就是原子字节块但跨两步的逻辑比如先读头部再读体需要调用方自己保证不被其他消费者打断最稳妥的办法是每个消费者进程独占一个队列或者用消息头里带完整数据的方案。3.3 主函数里的生产消费示例为了让这套代码能被直接验证我用两个独立进程演示。生产者每隔 1 毫秒写一条 128 字节的消息消费者同步读取并统计总数。// producer.cpp int main() { ShmFifo fifo; if (!fifo.init(/demo_shm_fifo, 4096 * 1024, true)) { fprintf(stderr, init failed\n); return 1; } char buf[128]; uint64_t counter 0; for (int i 0; i 100000; i) { memset(buf, 0, sizeof(buf)); snprintf(buf, sizeof(buf), msg-%llu, (unsigned long long)counter); if (!fifo.write(buf, sizeof(buf))) { fprintf(stderr, write failed at %llu\n, (unsigned long long)counter); break; } } return 0; } // consumer.cpp int main() { ShmFifo fifo; if (!fifo.init(/demo_shm_fifo, 4096 * 1024, false)) { fprintf(stderr, init failed\n); return 1; } char buf[128]; uint64_t received 0; while (received 100000) { if (fifo.read(buf, sizeof(buf))) { received; } } printf(received %llu messages\n, (unsigned long long)received); return 0; }/demo_shm_fifo是共享内存对象的全局名字所有进程通过同一个名字找到同一块内存。名字放在/dev/shm/下shm_open的第一个参数必须以/开头这是 POSIX 的要求。生产者和消费者的容量参数必须一致消费者init里的capacity实际不会生效因为createfalse时不会写头部真正生效的是创建进程写入的capacity值这一点我封装时没做校验如果两个进程传的容量不同消费者里拿到的还是生产者的值。信号量初值有一个细节要留意sem_empty初值是整个数据区容量 4 MBsem_full初值是 0。生产者写入 128 字节sem_empty减 128sem_full加 128。当生产速度远大于消费速度时sem_empty会降到 0sem_wait阻塞生产者的write停在那里等待消费者释放。这是一种自然背压机制——不会因为写太快把内存写爆也不会丢数据。如果希望生产者在阻塞时做别的事情而不是干等可以把sem_wait换成sem_trywait返回EAGAIN时跳过本次写入这在流控场景下有实际价值但要注意意味着丢消息需要按业务判断。4. 避坑排查共享内存 FIFO 的六类翻车现场4.1 进程残留、权限与信号量重名的坑【现象 1】第二次启动程序直接报shm_open返回ENOENT或者mmap成功但读到满屏乱码。原因上一次进程没有正常执行sem_unlink和共享内存清理就退出了共享内存对象还在/dev/shm里但信号量的初值已经被消费掉一部分。新进程创建时sem_open用了O_CREAT但因为名字相同拿到的还是旧的信号量里面sem_full的值可能是 5而缓冲区里的数据早就不在了。解决在创建路径里强制shm_unlink 三个sem_unlink然后再重新创建。shm_unlink只是删除名字引用不影响已经mmap的进程继续操作所以安全。我的习惯是把 unlink 放在 init 的 create 分支最前面顺序是先 unlink 三个信号量再 unlink 共享内存再创建。ftruncate之后还要memset整个头部区域把上次的残留magic清掉避免某个旧连接方还没退出时误判数据有效。【现象 2】两个进程在不同用户下跑消费者sem_open返回EACCES。原因生产者用0644权限创建共享内存对象文件所有者是启动用户。消费者用户不在 owner 组也没有写权限mmap时PROT_WRITE失败。解决共享内存和信号量两个对象都要设置权限。我一般统一设置0666并约定所有相关进程运行在同一个专用用户下。如果在系统层面有安全要求可以用setfacl给指定用户单独授权或者走 Unix socket 做一次握手后由特权进程代为打开。但最省事的做法是让部署脚本统一用sudo -u service_user拉起所有进程。【现象 3】信号量名字太长sem_open报ENAMETOOLONG。原因POSIX 信号量名字上限是NAME_MAX - 4通常是 251 字节。我的共享内存名/demo_shm_fifo没问题但有人会把共享内存名设计成/app/module/queue/data-pipeline这种 30 字符以上的路径形式加上_empty、_full后缀直接超限。解决信号量名字用独立的短别名不与共享内存名完全一致。比如共享内存叫/shm_data_pipeline_001信号量就叫/sf_001_em和/sf_001_fl用一定规则映射避免超长。这个问题隐蔽在部署后期一旦出现要改名字所有进程得一起升级成本比想象高。4.2 数据布局、单条过大与死锁类问题【现象 4】数据读出来前面几十个字节是乱的结构体字段被错位存取。原因我最初设计头部时用了柔性数组uint8_t data[]并且没关注结构体对齐。ShmFifoHeader里若增加一个uint64_t字段编译器会按 8 字节对齐sizeof(ShmFifoHeader)可能是 32 或 40而数据区的起始偏移算错了。更糟的是不同机器上sizeof不一样一个在 x86 上编一个在 ARM 上编偏移全错。解决头文件里强制#pragma pack(push, 1)并且不依赖sizeof计算偏移改用固定的offsetof或显式常量。我在中继一点代改进了数据区起始偏移写死为sizeof(ShmFifoHeader)但这个sizeof必须在编译期用静态断言固定比如static_assert(sizeof(ShmFifoHeader) 16)一旦有人改结构体导致大小变化编译直接报错比运行期乱码好排查得多。【现象 5】写入一条 2 MB 的数据程序卡死生产者消费者全都停住。原因容量是 4 MB看起来放得下 2 MB。问题出在环形缓冲区尾部如果write_pos已经走到 3 MB尾部只剩 1 MB而len2MB。代码里len tail判断为假会跨到头部写第二段。但头部的空间是 1 MB总容量够了。看起来没问题实际上我最初版本里tail计算错误用了cap - wpos却忘记加上头部已回绕部分导致tail虚高写入位置超出容量。解决严格按照环形公式处理并且用单测覆盖“写位置靠近缓冲区末尾”的回绕场景。最直接的办法是在write开头加断言(wpos len) % cap必须小于wpos与wpos len之间的所有位置实际上只要有len cap这个前置条件回绕后不会重合。第二个容易踩的坑是sem_empty初值是整个容量写 2 MB 时先扣掉 2 MB 空位回绕没有额外消耗所以逻辑没问题。如果你的队列还要支持单条接近容量上限的写建议容量至少是最大单条的 2 倍加 1 字节留出空位否则会造成写者永远等不到足够空间。【现象 6】多消费者进程同时read同一批数据收到重复消息或消息被拆开。原因sem_mutex只保护指针更新不保护两条read之间的业务逻辑。消费者 A 读了 4 字节长度头消费者 B 紧接着读了数据体两条消息错位。这个现象在单消费者模式下不存在一上多消费者就会出现。解决多消费者场景用“广播”或“互斥消费组”二选一。如果要多个进程都拿到同一份数据不能用 FIFO应该用发布订阅模型指针移位的设计就不适用了。如果只是多个消费者分担负载负载均衡要在外部加一把跨进程的分布式锁或者把队列拆成多个分片每个消费者固定消费一个分片。ShmFifo 的定位是单消费者模型多消费者接入需要在架构层面重新设计代码层面补不回来。另外一个跟死锁相关的隐蔽坑信号量的获取顺序。在异常分支持的情况下如果生产者在sem_wait(sem_empty)之后又在别处调用了sem_wait(sem_mutex)而消费者在sem_wait(sem_full)之后也调sem_wait(sem_mutex)两个路径不构成环路。但如果你后续新增了第二个队列做转发A 队列的持有者在等 B 队列的空位而 B 队列的持有者在等 A 队列的空位死锁就来了。解决方法是全局约定所有进程的信号量获取顺序必须一致按队列编号从小到大排列不破坏这个顺序就不会死锁。5. 验证与调优两个进程对跑实测吞吐与一致性5.1 一致性验证脚本校验和对比能编译不能说明正确必须跑一致性验证。我写了个双进程压测脚本来同时验证“不丢”“不重”“不错”。生产者和消费者各跑一个进程生产者生成 10 万个 128 字节的消息每条消息里放一个 uint64 序号和一个 CRC32 校验值。消费者读到后校验序号是否连续、CRC 是否匹配。# 终端 1启动消费者先等待共享内存创建 ./consumer_verify # 终端 2启动生产者创建共享内存并开始写入 ./producer_verify 100000 128 # 消费者退出时会输出 # received: 100000, checksum ok, sequence gap: 0这里的校验有两个关键点序号连续性用来验证不丢不重CRC 用来验证字节没有错位。跑一次只能证明基本路径没问题还要专门测回绕场景——让生产者一次性写入大量数据迫使write_pos越过缓冲区尾部至少两次。我是这么干的容量设成 1 MB单条消息 256 KB连续写入 100 条让write_pos反复回绕。这个用例在最初版本抓到了尾部拷贝错误后来修正后连续跑 24 小时没有异常。5.2 性能调优三板斧页对齐、批量写、超时禁调优环节我通常看三个地方。第一容量和映射要对齐到 4096 字节的整数倍。共享内存按页映射容量不对齐最后一页部分浪费而且memcpy时可能横跨两个页触发额外的缺页中断。容量 4 MB 很合适正好 1024 页。如果数据量大可以考虑MAP_HUGETLB用大页TLB miss 降低吞吐能再提升一截但要系统配置大页支持不是所有环境都允许。第二批量读写替代逐条读写。逐条 128 字节的sem_waitsem_post加锁开销很大我压测里每秒约 50 万条这已经不错了。但把消息合并成 8 KB 一块写吞吐还能翻倍。做法是在生产者侧攒批凑满 8 KB 或者超过 10 毫秒就 flush 一次。消费端相应地成块读再本地拆包。这个改动只是调用方式变化ShmFifo 的接口不用动write的len参数传 8 KB 即可。第三所有sem_wait加超时。生产者的sem_wait改成sem_timedwait并设置 100 毫秒超时超时后返回EAGAIN外层记一条告警日志。否则一次消费者崩溃生产者永远阻塞在sem_wait上整个链路假死日志里什么都看不到。sem_timedwait需要CLOCK_REALTIME的绝对时间取当前时间加超时值用clock_gettime(CLOCK_REALTIME)获取struct timespec ts; clock_gettime(CLOCK_REALTIME, ts); ts.tv_nsec 100 * 1000000; // 100ms if (ts.tv_nsec 1000000000) { ts.tv_sec 1; ts.tv_nsec - 1000000000; } int ret sem_timedwait(sem_empty_, ts); if (ret ! 0 errno ETIMEDOUT) { // 记录超时走降级路径 }tv_nsec溢出复位是常被忽略的细节不处理会在每秒钟附近时间点产生错误超时。这个坑我在某个凌晨排查告警时踩过现象是每整秒有一次超时误报。从那以后我每次写共享内存 FIFO都会强制走一遍三件事先验证回绕单测再跑 24 小时双进程压测最后把所有sem_wait全部换成带超时版本。这套习惯帮我挡住了不少线上事故虽然一开始多花半天时间但这半天每当遇到诡异问题时就值回来了。希望帮到你。本文还有配套的精品资源点击获取