Linux管道进程间通信：原理、选型与多进程日志收集实战

1. 为什么管道是进程间通信里最"土"却最稳的那把刀

管道这个东西，放在今天一堆消息队列、共享内存、RPC 框架中间，看起来确实土得掉渣——它连个消息边界都没有，两边只能传字节流，读的一方还得自己拆包。但你要是真在 Linux 上写过几年系统程序，就会发现一个反常识的现象：越是底层的、越是需要确定性的场景，最后拍板用的往往还是管道。shell 里那条竖线|每天都在跑，守护进程的父子进程之间传心跳，日志采集器把标准错误重定向给处理进程，甚至很多构建工具内部的任务调度，底层都是 pipe 加 fork 那套东西。

这篇文章聊的就是基于管道进行进程间通信这个话题。它解决的问题很具体：两个（或多个）进程之间怎么把数据从一个进程的地址空间挪到另一个进程里。注意是"挪"，不是"共享"——这是管道和共享内存最本质的分野。管道走的是内核缓冲区，数据从写端进内核、从内核出到读端，两个进程各自的地址空间互不可见，天然就带了一层隔离。

适合谁看？我把它分成三档。第一档是刚学完 fork 和文件描述符、正准备动手写第一个双进程程序的同学，你需要一个能跑通、能看懂、还能改的最小完整例子。第二档是写过一些脚本、想搞清楚 shell 里cmd1 | cmd2到底在内核里发生了什么的人，理解之后你在写自动化脚本时对缓冲、阻塞、退出码的判断会准很多。第三档是做系统工具、容器、编辑器插件这类需要在进程间做轻量协调的开发者，管道往往是成本最低的候选方案，但坑也不少，后面我会把踩过的那些都摊开讲。

先给个结论性的认知：管道的核心价值不在"能传数据"，而在它把同步问题一起解决了。你去用共享内存，得自己配信号量；你去用消息队列，得自己管理消息生命周期；而管道自带流控——内核缓冲区满了，写端自动阻塞；缓冲区空了，读端自动阻塞。这个"自动"两个字，省掉的是大量状态机的代码。代价也很明确：只能单向传，只能传字节流，只能用于有亲缘关系的进程（匿名管道），以及默认只有 64KB 的缓冲区。这些限制不是缺陷，是设计取舍，理解取舍才知道什么时候该用它、什么时候该换别的。

下文我会从整体设计思路讲到具体实操，中间穿插我自己写过的一个多进程日志收集的小例子，最后把常见的坑整理成一张速查表。所有代码都能直接编译运行，所有参数都给了计算依据，你照着抄作业就行。

2. 整体设计与方案选型：匿名管道和命名管道到底怎么挑

2.1 先搞清楚管道在内核里是什么东西

动手之前必须建立一个正确的心理模型，不然你写的代码大概率是靠猜在跑。管道在 Linux 内核里就是一块环形缓冲区，用pipe()系统调用创建时会同时返回两个文件描述符，fd[0]固定是读端，fd[1]固定是写端，这个顺序千万不能记反——我见过不止一个人把两个参数传反，然后程序卡死在第一个 read 上，debug 半天。

这块缓冲区默认大小是 64KB，也就是 65536 字节，但这个值不是硬编码的常量，它受/proc/sys/fs/pipe-max-size限制，而实际的管道容量可以用fcntl(fd, F_GETPIPE_SZ)查出来。在大部分发行版上你查到的数字会是 65536，但在某些内核配置下可能是 4096 的整数倍但不到 64K，比如 8192。这个数字直接决定了你的程序在什么时候会阻塞，所以写代码前最好实测一下，别信网上的固定值。

管道有几个派生概念，容易混，我列个表说清楚：

这个表建议你记下来，尤其是最后一列——很多人一上来就用 socketpair 做双向通信，其实如果只是单方向传数据，用普通的pipe()更轻，连接建立的开销几乎为零。

2.2 匿名管道 vs 命名管道：选型的三个判断维度

第一个维度是进程关系。匿名管道只能通过 fork 继承文件描述符来共享，这意味着通信双方必须有共同的祖先。如果你要通信的是两个完全独立启动的程序，比如一个独立部署的采集进程和一个独立部署的处理进程，那就只能用命名管道。命名管道本质是一个特殊文件，在文件系统里有路径，任何有权限的进程都能 open 它。

第二个维度是方向性。匿名管道是单向的，要双向就得建两个。命名管道也一样，一个 FIFO 打开之后就只能单向数据流（虽然技术上 open 时可以同时读写，但那样行为很怪，几乎没人这么用）。如果你的通信模型是"请求-响应"，也就是 A 发请求、B 回结果，那用一对管道是标准做法，或者直接上 socketpair。这里我要强调一点：别为了省一个 fd 去自己跟自己玩双向 FIFO，语义混乱，出了 bug 极难查。

第三个维度是生命周期管理。匿名管道的 fd 一旦所有引用都关掉，内核缓冲区就释放了，不需要你手动清理。命名管道不一样，它在文件系统里留了个 inode，程序退出了文件还在，下次再 open 可能读到上一次残留的数据（虽然 FIFO 本身不缓存数据，但如果你的程序在 open 后没读干净，确实会有意外）。所以用 FIFO 的时候，我的经验是一定要配一个明确的清理策略：要么在程序退出前unlink，要么在启动时先判断文件是否已存在并做相应处理。

2.3 为什么不用共享内存和消息队列

这个问题我在技术分享里被问过很多次。直接说结论：判断标准是有没有"同步需求"和"数据规模"。

共享内存的优势是零拷贝，数据量大的时候性能差距能到十倍以上，因为管道要走两次内核拷贝（写时用户态到内核态，读时内核态到用户态）。但共享内存本身完全不带同步机制，你要自己实现互斥和条件变量，一不小心就是竞态。我在一个数据采集项目里做过对比：单次传输 1MB 数据，共享内存加信号量大约是 30 微秒级别，管道是 200 微秒级别。差 6 倍多，但那个项目的传输频率只有每秒几十次，这点差距完全可以忽略，而管道的代码量只有共享内存方案的三分之一。

消息队列是内核持久化的，支持多对多，有消息边界，这些特性在需要解耦的系统里很有价值。但它的开销比管道大，而且消息队列在内核里占用的资源是有限的（受msgmni等参数限制），普通的小数据传递用它属于杀鸡用牛刀。

所以我的选型经验是：凡是单个消息小于 64KB、传输频率不算极端、双方有亲缘关系或者可以接受 FIFO 这套机制的，优先用管道。数据量大到几十 MB、或者需要真正的多生产者多消费者，再考虑共享内存或消息队列。这个原则在我过去几年写的代码里几乎没有踩过坑。

2.4 整体架构设计：一个具体的通信模型

为了让后面的实操有落点，我设计一个真实场景：多进程日志收集。主进程负责接收业务数据并计算指标，三个工作进程负责把各自的运行状态（处理条数、错误数、耗时）上报给主进程，主进程汇总后统一输出。

这个模型的通信需求是：三个工作进程 → 主进程，单向的、小数据量的、周期性的。三个工作进程是主进程 fork 出来的，有明显的亲缘关系。所以我选三个匿名管道，每个工作进程独占一个写端，主进程持有三个读端。

关键的设计点在于文件描述符的关闭策略。这里有个经典错误：主进程 fork 完三个子进程后，每个子进程手里都持有全部六个 fd（三个读端三个写端），如果不在子进程里把不需要的 fd 关掉，那么当某一个工作进程退出时，主进程是感知不到的，因为写端还被其他子进程持有，read 不会返回 0。这个问题的表现是：你明明 kill 了一个工作进程，主进程却没有发现，汇总数据里少了它的记录，但程序也不报错。我在第一次写这个模型时就被这个坑蹲了差不多两个小时。

正确的做法是：fork 之后，每个子进程立刻把除自己那个写端以外的所有管道 fd 全部 close。主进程则把所有写端都 close，只保留读端。这样任何一个子进程退出，它的写端自动关闭，主进程对应的 read 就会返回 0，从而知道这个通道结束了。这个原则叫"最小持有"，是所有多进程 fd 管理的基础。

3. 核心细节解析：从 pipe 调用到读端行为的所有关键点

3.1 pipe() 的返回值和错误处理不能偷懒

pipe()的原型是int pipe(int pipefd[2])，成功返回 0，失败返回 -1 并设置 errno。常见错误码里，EMFILE是进程打开的文件描述符超过上限（默认通常是 1024，可以用ulimit -n看），ENFILE是系统级的总 fd 数超了，EFAULT是传入的地址非法。

实际的坑在于，很多人写示例代码的时候只写pipe(fd);，连返回值都不判。在演示程序里没问题，但生产代码里如果你在循环里创建管道、又没判错，很快就把 fd 耗尽了，然后程序进入一种莫名其妙的状态——之后所有的 open、socket 全部失败，报的错完全指向别处。

我的习惯是封装一个函数：

#include <unistd.h> #include <stdio.h> #include <string.h> #include <errno.h> int safe_pipe(int fd[2]) { if (pipe(fd) == -1) { fprintf(stderr, "pipe failed: %s\n", strerror(errno)); return -1; } return 0; }

一行strerror(errno)能在排障时省掉大量时间。另外提醒一点：pipe()返回的两个 fd 默认是没有FD_CLOEXEC标志的，也就是说如果你之后调了exec，这两个 fd 会被继承到新程序里。这在 shell 管道里是必须的行为（不然数据传不过去），但在你自己写程序时，如果没有 exec 需求，建议立刻设上 CLOEXEC：

fcntl(fd[0], F_SETFD, FD_CLOEXEC); fcntl(fd[1], F_SETFD, FD_CLOEXEC);

或者直接用 Linux 专有的pipe2(fd, O_CLOEXEC)，一步到位，还能顺便设O_NONBLOCK。pipe2是 2.6.27 之后的内核才有的，现在的发行版都支持，所以我个人的新代码基本都用pipe2。

3.2 读写的语义细节：返回值、短读和部分写

这是管道最容易出 bug 的地方，必须掰开了说。

写端write(fd[1], buf, n)的行为：只要还有至少 1 字节的空间，内核就会尽量多写，返回实际写入的字节数。注意是"尽量多写"，不是"一定写 n 字节"。如果缓冲区剩余空间不够 n 字节，可能只写了一部分，返回一个小于 n 的值，这叫部分写。如果你写的时候缓冲区已经满了，那就直接阻塞，直到有空间为止（阻塞模式下）。

读端read(fd[0], buf, n)的行为：如果缓冲区里有数据，就返回当前可读的字节数（最多 n 字节），不一定是你上次写进去的完整长度。如果缓冲区为空且还有写端开着，阻塞等待。如果所有写端都关闭了，返回 0，表示 EOF。如果被信号打断，返回 -1 且 errno 是EINTR。

这里的关键认知是：管道不保证消息边界。你写两次，一次 10 字节一次 20 字节，读端可能一次读出 30 字节，也可能分三次读出 5、15、10 字节。所有基于管道做结构化数据传递的代码，都必须自己实现一套定界协议。

最常用的定界方案有两种。第一种是长度前缀：每个消息前面加 4 字节表示后续 payload 的长度，读端先读 4 字节，解析出长度，再循环读到够为止。第二种是分隔符：以换行符之类的特殊字符作为消息结束标记，适合文本协议。长度前缀更严谨，二进制安全，我在实际项目里一律用它；分隔符方案写起来快，但一旦数据里本身包含了分隔符就会崩，需要转义，反而麻烦。

读端还有个非常常见的错误：以为read会读到指定长度。比如你写read(fd, buf, 1024)，以为一次读到 1024 字节，结果实际只读了 100 字节，后面 900 字节是你栈上的垃圾数据。这是 C 语言初学者最经典的 bug 之一。正确的做法永远是检查返回值，然后循环读直到读够：

ssize_t read_full(int fd, void *buf, size_t need) { size_t done = 0; char *p = (char *)buf; while (done < need) { ssize_t n = read(fd, p + done, need - done); if (n > 0) { done += n; } else if (n == 0) { return (ssize_t)done; /* EOF，返回已读到的部分 */ } else { if (errno == EINTR) continue; /* 被信号打断，重试 */ return -1; } } return (ssize_t)done; }

这几行代码里有三个细节值得单独说。done用 size_t 无符号类型，避免负数累加；EINTR必须重试，否则信号一来你的读操作就废了；返回 0 的情况要区分——可能是正常 EOF，也可能是对方的写端关了，调用方必须能区分"读满了"和"对端关闭了"这两种情况，前者返回的 done 等于 need，后者小于 need。

3.3 阻塞与非阻塞：什么时候该把管道设成非阻塞

默认情况下管道是阻塞的。阻塞语义在"生产者-消费者"这种纯粹的流水线模型里是完美的：写满就等，读完就等，CPU 一点都不浪费。但它有两个致命场景。

场景一：多路复用。主进程要同时从三个管道读数据，如果用阻塞 read，读第一个的时候后两个的数据就只能干等着，延迟被拉大。这种情况要么用select/poll/epoll做多路复用（推荐，ReadFds 里放三个 fd，谁有数据读谁），要么把三个都设成非阻塞轮询（不推荐，CPU 空转）。我在日志收集那个例子里用的是poll，因为三个 fd 数量少，poll 的开销完全可接受，代码也比 epoll 简单。

场景二：不能卡住的主循环。比如你在写一个交互式程序，主线程要同时处理用户输入和进程间消息，阻塞 read 会把整个 UI 冻住。这时候必须用非阻塞 fd 或者放到独立线程里。

设置非阻塞有两种方式：

/* 方式一：pipe2 一步到位 */ int fd[2]; pipe2(fd, O_NONBLOCK | O_CLOEXEC); /* 方式二：fcntl 事后设置 */ int flags = fcntl(fd[0], F_GETFL); fcntl(fd[0], F_SETFL, flags | O_NONBLOCK);

非阻塞模式下，read 在没数据时返回 -1 且 errno 为EAGAIN（或EWOULDBLOCK，两者在 Linux 上是同一个值），write 在空间不足时也返回 -1 加EAGAIN。注意EAGAIN不是错误，是正常的"请稍后再试"，你的代码里必须把它和其他错误区别对待，否则会误判。

我一直觉得有个细节容易被忽略：非阻塞的 write 在做部分写的时候，你没法知道到底写了多少，只能靠返回值。如果你的消息是 100 字节，第一次只写了 60，剩下 40 你必须缓存起来等下次可写时继续写。这就需要一个应用层的发送队列，复杂度立刻上来了。所以我给的建议是：只有在你确实需要多路复用、或者主循环不能被卡住的时候才用非阻塞，否则老老实实用阻塞。阻塞模型配合 select/poll，在很多场景下比纯非阻塞更简单也更可靠。

3.4 SIGPIPE 信号：写一个已经关闭的管道会发生什么

这个知识点我单独拎出来讲，因为它是管道编程里最"隐蔽"的一类问题。

当你在一个所有读端都已关闭的管道上执行 write 时，内核会先给当前进程发一个SIGPIPE信号。这个信号的默认动作是终止进程。注意，进程是被信号杀掉的，不是 write 返回错误，代码里你根本走不到错误处理那一行，程序直接没了。

shell 管道这个现象最典型：yes | head -n 1，head读一行就退出，关闭了读端，yes继续写，收到 SIGPIPE 就死了。这是符合预期的行为，不需要处理。

但在自己的程序里，如果你没意识到这一点，就会遇到"程序莫名退出、没有 core dump、日志里什么都没有"的情况。我第一次遇到时愣了半小时，后来用strace才看到是被 SIGPIPE 干掉的。

有两种处理方式。第一种是忽略信号，让 write 返回EPIPE错误，由代码处理：

signal(SIGPIPE, SIG_IGN);

第二种是在具体的写操作前用send的MSG_NOSIGNAL标志——但那是 socket 的接口，管道用不了，所以管道只能走第一种。

忽略掉之后，write 会返回 -1，errno 为 EPIPE。这时你应该判断：是对端真的关闭了（正常情况），还是逻辑错误（不该关的时候关了）。前者就清理退出，后者要打日志。我的习惯是，只要程序里用到管道，就在main开头直接把 SIGPIPE 设成忽略，然后把 EPIPE 当成一个明确的错误码来处理，避免进程被信号悄无声息地干掉。

3.5 缓冲区容量与原子性：PIPE_BUF 那 4096 字节的门槛

这里有个宏叫PIPE_BUF，在 Linux 上值是 4096。它定义了一个非常重要的边界：小于等于 PIPE_BUF 的写操作是原子的，大于的不保证。

"原子"在这个语境下的意思是：多个进程往同一个管道的写端写数据时，如果每次写入都不超过 PIPE_BUF 字节，那么这些写入不会互相穿插，每个 write 的内容是连续的一块。超过这个值，就可能出现 A 写的 10KB 和 B 写的 10KB 交错在一起，读端读到的数据是乱的。

这个特性在"多个子进程往同一个管道写日志"的场景里是刚需。我的日志收集例子如果改成三个工作进程共享一个管道写端，那么每条日志的大小就必须控制在 4096 字节以内，否则日志会串。稳妥的做法是：要么保证单条消息小于 PIPE_BUF 并依赖原子性，要么每条消息配一个独立的管道或加锁。我一般选前者，因为日志这种场景单条信息很少超过 4KB，实在超了就截断并标注。

顺带说下容量查询的方式：

# 查看系统允许的最大管道容量 cat /proc/sys/fs/pipe-max-size # 代码里查询某个管道的实际容量 fcntl(fd[0], F_GETPIPE_SZ)

Linux 还允许用F_SETPIPE_SZ调整管道容量（需要足够的权限，且不能超过 pipe-max-size），最小是一页（4096），最大默认 1MB。调大有好处——减少阻塞次数；也有坏处——内存占用上升，且数据的延迟变高（数据积在缓冲里还没被读走）。我一般不动这个值，除非明确测出阻塞是瓶颈。

4. 实操过程：从零写一个能跑的多进程日志收集

4.1 环境确认和编译参数

先把环境说清楚。我用的是一台普通的 x86_64 机器，内核版本 5.15，glibc 2.35，gcc 11。你用什么发行版影响不大，管道这套接口 POSIX 标准化得很彻底，唯一需要注意的是pipe2在极老的内核上没有，那种情况退回pipe加fcntl就行。

编译命令我建议加上几个警告选项，管道相关的 bug 很多都能被编译器提前抓住：

gcc -Wall -Wextra -O2 -o logcollect logcollect.c

-Wall -Wextra会提示未使用的变量、返回值被忽略之类的常见问题——比如你写了read(...)却没接收返回值，它会警告，这在管道代码里非常有用。

4.2 第一步：规划消息格式和数据通道

消息格式我定成定长的二进制结构，这样解析最简单，也不用处理序列化：

#include <stdint.h> #define MSG_MAGIC 0x4C4F4701 typedef struct { uint32_t magic; /* 4 字节，固定魔数，用于校验 */ uint32_t worker_id; /* 4 字节，工作进程编号 */ uint64_t count; /* 8 字节，已处理条数 */ uint64_t errors; /* 8 字节，错误条数 */ uint64_t elapsed_ms; /* 8 字节，累计耗时 */ } __attribute__((packed)) WorkerStat; #define STAT_SIZE ((size_t)sizeof(WorkerStat))

结构体总大小是 4+4+8+8+8 = 32 字节，远小于 PIPE_BUF，原子性有保障。__attribute__((packed))是为了避免编译器在字段之间插入对齐填充，保证结构体大小在我这里是精确的 32 字节，跨机器传输时也稳。不过要提醒一句：packed 结构体在访问未对齐字段时在某些架构（比如老 ARM）上会慢，甚至触发异常。如果你的程序只在 x86 上跑，无所谓；如果要跨架构，老老实实按字段序列化更稳。

4.3 第二步：主进程创建管道并 fork 工作进程

核心代码我完整贴出来，每一行都标了为什么：

#include <stdio.h> #include <stdlib.h> #include <unistd.h> #include <fcntl.h> #include <poll.h> #include <signal.h> #include <string.h> #include <errno.h> #include <stdint.h> #include <sys/wait.h> #include <time.h> #define WORKER_NUM 3 typedef struct { uint32_t magic; uint32_t worker_id; uint64_t count; uint64_t errors; uint64_t elapsed_ms; } __attribute__((packed)) WorkerStat; int main(void) { /* 关键一：忽略 SIGPIPE，避免写端关闭时进程被信号杀死 */ signal(SIGPIPE, SIG_IGN); int pipes[WORKER_NUM][2]; pid_t pids[WORKER_NUM]; for (int i = 0; i < WORKER_NUM; i++) { if (pipe2(pipes[i], O_CLOEXEC) == -1) { fprintf(stderr, "pipe2 failed: %s\n", strerror(errno)); return 1; } } for (int i = 0; i < WORKER_NUM; i++) { pid_t pid = fork(); if (pid < 0) { fprintf(stderr, "fork failed: %s\n", strerror(errno)); return 1; } if (pid == 0) { /* 子进程：关掉所有不属于自己的 fd */ for (int j = 0; j < WORKER_NUM; j++) { close(pipes[j][0]); /* 读端一律不要 */ if (j != i) close(pipes[j][1]); /* 别人的写端也不要 */ } worker_loop(i, pipes[i][1]); close(pipes[i][1]); /* 正常结束时显式关闭，触发父进程 EOF */ _exit(0); } pids[i] = pid; } /* 父进程：关掉所有写端，只留读端 */ for (int i = 0; i < WORKER_NUM; i++) { close(pipes[i][1]); } /* 主干：多路复用收集 */ int alive = WORKER_NUM; while (alive > 0) { struct pollfd pfds[WORKER_NUM]; int n = 0; for (int i = 0; i < WORKER_NUM; i++) { if (pipes[i][0] >= 0) { pfds[n].fd = pipes[i][0]; pfds[n].events = POLLIN; pfds[n].revents = 0; n++; } } int ready = poll(pfds, n, 1000); if (ready < 0) { if (errno == EINTR) continue; perror("poll"); break; } if (ready == 0) continue; /* 超时，下一轮 */ for (int k = 0; k < n; k++) { if (!(pfds[k].revents & (POLLIN | POLLHUP))) continue; WorkerStat st; ssize_t got = read_full(pfds[k].fd, &st, sizeof(st)); if (got == 0) { /* EOF：对端关闭 */ close(pfds[k].fd); for (int i = 0; i < WORKER_NUM; i++) { if (pipes[i][0] == pfds[k].fd) pipes[i][0] = -1; } alive--; continue; } if (got < 0) { perror("read"); continue; } if (got < (ssize_t)sizeof(st)) { fprintf(stderr, "short read %zd bytes\n", got); continue; } if (st.magic != 0x4C4F4701) { fprintf(stderr, "bad magic from worker %u\n", st.worker_id); continue; } /* 汇总逻辑 */ accumulate(&st); } } ... }

写这段代码时我修过好几个问题，这里挑最重要的三个讲。

第一个问题是poll的revents不能只看POLLIN。对端关闭时内核会设置POLLHUP，有时候数据已经读到一半了对端挂了，你可能同时收到 POLLIN 和 POLLHUP。如果只判断 POLLIN，会漏掉关闭事件，导致循环卡住。所以判断条件写成(revents & (POLLIN | POLLHUP))，然后靠 read 返回 0 来识别真正的结束。

第二个问题是关掉 fd 之后必须把数组里对应的位置置为 -1，否则下一轮 poll 还会拿这个已经关闭的 fd 去注册，poll 会返回POLLNVAL错误，或者更糟——如果你在别处又打开了文件，fd 号被复用了，poll 就会去监听一个完全不相干的文件，行为极其诡异。fd 复用这个坑我踩过一次，排查花了大半天。

第三个问题是必须循环调用read_full而不是一次 read。这是前面讲过的短读问题，虽然 32 字节的消息在 64KB 缓冲区里几乎不可能被切断，但如果你在写端连续写了很多消息，读端一次 read 的数量可能超过一条消息的长度，或者恰好读到一半。用 read_full 保证每次读满一条完整消息，逻辑才干净。

4.4 第三步：工作进程的活动循环和退出处理

工作进程的代码相对简单，但有几个细节值得说：

static void worker_loop(int id, int wfd) { unsigned seed = (unsigned)(time(NULL) ^ (getpid() << 8)); for (int round = 0; round < 10; round++) { /* 模拟业务处理 */ int batch = 100 + rand_r(&seed) % 400; int errors = (rand_r(&seed) % 20 == 0) ? 1 : 0; uint64_t elapsed = 5 + rand_r(&seed) % 50; WorkerStat st; st.magic = 0x4C4F4701; st.worker_id = (uint32_t)id; st.count = (uint64_t)batch; st.errors = (uint64_t)errors; st.elapsed_ms = elapsed; size_t done = 0; const char *p = (const char *)&st; while (done < sizeof(st)) { ssize_t n = write(wfd, p + done, sizeof(st) - done); if (n < 0) { if (errno == EINTR) continue; fprintf(stderr, "worker %d write: %s\n", id, strerror(errno)); return; } done += (size_t)n; } usleep(200000); /* 模拟周期性上报，200ms 一次 */ } }

写端用了循环处理部分写，虽然 32 字节远小于 PIPE_BUF，正常情况下一次就写完了，但在阻塞模式下如果缓冲区恰好接近满，理论上仍可能只写一部分。写循环的成本几乎为零，加上更稳。

usleep(200000)是为了模拟真实业务里周期性上报的节奏，同时给你一个观察点——如果主进程处理不及，写端会被阻塞，你能从 10 轮的总耗时上看出背压。

4.5 编译运行与验证方法

编译后跑起来，输出类似这样：

worker 0: count=1204 errors=1 elapsed=178 worker 1: count=1810 errors=0 elapsed=203 worker 2: count=2350 errors=2 elapsed=167 total count=5364, error rate=0.06%

验证管道真的生效，我习惯用三个手段。第一是strace -f -e trace=pipe2,read,write,close ./logcollect 2>&1 | head -50，能直接看到每个进程的 pipe2、write、read 系统调用序列，一目了然。第二是ls -l /proc/<pid>/fd，在程序运行期间查看某个工作进程持有的 fd，确认它真的只有自己的写端。第三是人为制造一次对端提前退出：在工作进程里改成只跑 2 轮就_exit，看主进程是否正常汇总完退出、有没有卡死。

关于性能，我实测这个模型在三个进程各上报 10 次的情况下，端到端延迟在 1 毫秒以内，CPU 占用几乎为 0（poll 阻塞等待，没有空转）。如果我把上报间隔改成 10 毫秒，总消耗也不到 5% 的单核。这个量级说明管道在低频通信场景下的开销完全可以忽略，你不需要为了性能去折腾更复杂的东西。

5. 排查实录：管道相关问题的速查表和踩坑经验

5.1 常见问题速查表

表里每一条我都在实际工作中遇到过至少一次，排查方向是"先看现象，再定位层，最后验证"。

5.2 那个让我印象最深的 fork 后 fd 泄漏

说个具体的。有一年我写一个任务分发器，主进程 fork 出 N 个 worker，worker 通过管道回传进度。代码写完后测试都正常，上线后跑一段时间就出现问题：某些 worker 崩溃后，主进程对应的 read 永远不返回，整个汇总卡住。

原因就是我前面强调过的 fd 最小持有原则——worker 的写端在主进程和所有其他 worker 里都有一份拷贝。某个 worker 崩溃后，它自己的写端关了，但其他 worker 还持有那一份，所以主进程的 read 永远等不到 EOF。修复就是在 fork 之后、进入业务循环之前，把所有不属于自己的 fd 全部关掉。

改完之后我加了一条防御性断言，在 worker 启动时遍历/proc/self/fd，检查持有的 fd 数量是否等于预期。这个检查在生产环境里也许略重，但在开发和测试环境里非常值——它能立刻发现 fd 泄漏，而不是等到线上某天出事。

另一个坑是close之后继续用同一个 fd 号。因为 Linux 分配 fd 是最小可用优先，你一 close，马上 open 一个新文件，很可能就拿到同一个 fd 号。如果你的代码里某个地方还存着旧 fd 的副本，就会去操作了一个完全不同的文件。这类 bug 的表现是"数据写到了不该写的地方"，排查难度极高。我的做法是：close 之后立刻把变量赋 -1，并且在所有使用点前判断if (fd < 0) continue;。

5.3 性能相关的两个实测结论

第一个结论：缓冲区大小对吞吐的影响主要在大量小消息场景。我做过一组对照，写 100 万条 32 字节的消息，用默认 64KB 缓冲区，吞吐大约是每秒 280 万条；把它调到 1MB，吞吐升到每秒 420 万条，提升 50%。原因是减少了两边进程的上下文切换次数。但注意，如果你每条消息处理时间比较长（比如几百微秒），这个差距会被业务时间掩盖，调大缓冲区没意义。

第二个结论：poll 的 fd 数量在几十个以内时，开销可以忽略。我测过 3 个、16 个、64 个 fd 的 poll 循环，单次轮询耗时分别在 1 微秒、3 微秒、6 微秒量级。到几百个 fd 的时候 epoll 才有明显优势。所以不要一看 fd 多了就上 epoll，先看实际数量级。我那个日志收集器三类 fd，用 poll 是对的，代码也短。

5.4 几条不写在文档里的实操心得

心得一：给每个管道起个名字，在日志里带上。管道没有名字，出问题时你只知道"某个 fd 卡住了"。我的做法是在定义管道数组的时候同时维护一个字符串表，日志里输出pipe[worker_2]这样的标识，排查效率提升巨大。

心得二：调试阶段把缓冲区调小。把F_SETPIPE_SZ设成 4096（一页），能让满缓冲和背压的问题更早暴露。生产环境再用默认值。这个技巧帮我提前发现过好几次写端阻塞导致死锁的代码路径。

心得三：永远假设对端随时会死。管道通信里对端崩溃是常态而不是异常。你的读端要有 EOF 处理，写端要有 EPIPE 处理，中间的数据要能容忍不完整（比如收到半个结构体就丢弃整条）。我见过太多代码把对端消失当成"不该发生的错误"，然后整个程序就僵在那里。

心得四：别在信号处理函数里操作管道。这是个经典陷阱。信号处理函数里能安全调用的函数非常有限（只有 async-signal-safe 的那一批），read、write、close虽然是 async-signal-safe 的，但你的业务逻辑不是。正确的模式是在信号处理函数里设一个volatile sig_atomic_t标志，主循环里检查这个标志再做事。我在早期代码里直接在 SIGCHLD 处理函数里 close 管道，结果偶发死锁，查了很久才定位到信号上下文里调用了不安全的操作。

心得五：父子进程共享同一个管道的读端要慎用。如果你 fork 出多个子进程，它们共享同一个管道的读端，那么数据会被"抢"走——哪个子进程先 read 就归谁，其他进程拿不到。这在需要工作分摊的场景下是有用的负载均衡，但在需要每个子进程都看到所有消息的场景下就是灾难。这时候要么每个子进程一条独立管道，要么用别的机制。我在写一个配置下发功能时就踩了这个坑，后来改成一进程一管道，问题立刻消失。

心得六：结构体传输要防同源不同编。前面用了 packed 结构体，好处是大小固定，但前提是双方都是同一份代码、同一个编译器设置、同一种字节序。如果你写端是 C 程序、读端是 Python 脚本，直接解析结构体就得手动处理对齐和字节序。跨语言场景我的建议是老老实实用sizeof加上明确的字节布局（按小端序逐字段写入），别依赖编译器的内存布局。这一点在实际项目里非常容易被忽略，尤其是在做多语言混合架构的时候。