去年年底帮朋友调一个数据处理程序，他用了最朴素的思路：每来一批文件，就在 shell 里循环 fork 子进程去处理。结果机器负载忽高忽低，任务一多甚至会把句柄数打到上限，进程创建的开销比干活本身还大。后来我给他换成了基于匿名管道的进程池，代码量没多多少，但吞吐和稳定性完全不一样了。这篇就把整个思路和可运行代码完整记录下来。
这个方案的核心思路是：在 Linux 系统上，预先 fork 出固定数量的子进程，父进程通过匿名管道把任务描述符写下去，子进程处理完后通过另一根管道把结果回传。这样一来，"创建进程"这个昂贵的操作只发生一次，后续都是轻量的管道读写，进程间通信的成本被压缩到了极致。如果你想自己实现一个简单的进程池，或者想彻底搞懂匿名管道在多进程协作里的实际用法，这篇文章可以直接参考。
1. 为什么是这个组合：进程池与匿名管道的契合点
1.1 多次 fork 的成本压垮了简单方案
很多 Linux 初学者会觉得 fork 很轻量，毕竟内核有写时复制技术，子进程不会真的完整复制一份父进程地址空间。但实测下来，一次 fork 的开销并不只是"复制页表"这么简单：
- 内核要创建新的 task_struct，分配 pid、挂载到调度队列；
- 要复制父进程的 mm_struct、文件描述符表、信号处理表；
- 如果父进程内存占用大，即使有 COW，页表本身的复制也不便宜；
- 如果后续还要 exec 加载新程序，代价更高：动态链接器要重新加载所有 .so，库映射、重定位全部重来。
我之前简单测过，裸 fork 一万次大概需要几百毫秒到一秒，这看起来不夸张，但它挤占了 CPU 时间片、拉高了平均负载。更麻烦的是，如果每个任务只运行几毫秒，那进程创建销毁的开销占比会非常高，系统大部分时间都在"造人"和"收尸"，而不是真正干活。
进程池的思路很简单：提前创建好 N 个进程，有任务就分发下去，没任务就让它们阻塞等待。这样把"频繁创建进程"变成"复用空闲进程"，系统负载自然平稳了。
1.2 匿名管道相比其他 IPC 方式好在哪
Linux 下进程间通信的方式很多：共享内存、消息队列、信号、socketpair、信号量等。为什么这个场景我选了匿名管道？因为它和 fork 配合得最自然。
关键点在于：pipe() 创建的两个文件描述符，在 fork 之后被子进程自动继承。也就是说，父进程只需要在 fork 之前创建好管道，父子进程天然共享这对读写端，不需要额外的名字、key、路径，也不需要连接建立过程。而共享内存虽然传输大块数据快，但需要自己处理同步，用信号量或锁来保证互斥，代码复杂度直接拉高；消息队列又涉及内核对象的管理和清理，协议也偏重。
从语义上看，管道是流式的、字节序严格 FIFO 的，天然适合"父进程下发指令、子进程回传结果"这种一问一答的模式。任务消息通常很小（几十个字节），管道的吞吐完全够用，内核还会在每次 write 时对不超过 PIPE_BUF 的数据保证原子性，这给消息协议带来了很大的方便。
当然，匿名管道不是万能的，它只能用于有亲缘关系的进程之间。我们的子进程全部是父进程 fork 出来的，正好满足这个前提，也规避了命名管道需要文件系统路径、可能有残留文件的问题。
2. 进程池的整体结构与通信协议设计
2.1 数据流设计：任务管道和结果管道分离
匿名管道是单工的，数据只能从写端流向读端。如果父进程要下发任务、子进程要回传结果，至少需要两根管道。我这里是给每个子进程单独分配两根管道，而不是让所有子进程共享一对：
- 任务管道：父进程持有写端，子进程持有读端。父进程通过它下发任务消息。
- 结果管道：子进程持有写端，父进程持有读端。子进程通过它回传执行结果。
为什么不共用一个管道？一个简单的版本里，所有子进程共用一根任务管道也没问题，多写者写管道时只要消息不超过 PIPE_BUF 且大小固定，写入本身不会交错。但结果管道如果共用，父进程读完结果后无法判断是哪位 worker 空闲，也就不知道该给谁派发新任务。每个 worker 独立一对管道，父进程就能根据"哪个结果管道可读"精确定位到空闲 worker，实现了真正的事件驱动调度。
这种设计下的数据流如下图所示（文字描述）：父进程任务队列 → 任务管道 → 子进程 A/B/C → 结果管道 → 父进程事件循环。整体是两条单向数据流交叉成环，但每个方向上访问独立，不会出现读写竞争。
2.2 消息协议与核心数据结构
管道本质是字节流，没有消息边界，所以通信双方必须约定好消息格式。我使用固定大小的结构体作为消息单元，避免在管道里做拆包解析：
#define CMD_EXIT 0 #define CMD_TASK 1 #define CMD_DONE 2 typedef struct { int cmd; // CMD_EXIT / CMD_TASK / CMD_DONE int task_id; // 任务编号，用于结果关联 int arg; // 任务参数，也可以承载结果值 } TaskMsg;固定 12 字节的消息，小于 PIPE_BUF 的 4096 字节，因此每次 write 都是原子的，read 端只要每次读取 sizeof(TaskMsg) 字节，就一定能拿到一条完整消息。这里有个常见误区：管道是字节流，不是报文队列。如果消息长度可变，就必须自己加上长度头或者分隔符。固定结构体是最省事的方案。
每个 worker 的状态用如下结构保存：
typedef struct { int task_wfd; // 父进程 -> 子进程：任务管道写端 int result_rfd; // 子进程 -> 父进程：结果管道读端 pid_t pid; // 子进程 pid int busy; // 1 表示当前有任务在执行，0 表示空闲 } Worker;task_wfd 是父进程视角的写端，result_rfd 是父进程视角的读端，这两个文件描述符就是父进程操作该 worker 的全部"遥控器"。
2.3 fork 后的文件描述符裁剪
父进程为每个 worker 创建两根管道后 fork，这里最容易被忽略的是：fork 出来的子进程会继承父进程当前所有的文件描述符，包括其他 worker 的管道写端和读端。如果不做处理，子进程手里会握着所有管道的所有端点，后果是灾难性的：
- 任务管道永远不会产生 EOF，因为即使父进程关闭了写端，子进程自己还握着别的 worker 的写端；
- 结果管道同理，父进程无法通过 read 返回 0 感知子进程退出；
- 每个管道都多了一堆无用的 fd，浪费内核资源。
所以每次 fork 后，子进程要立刻关闭自己不该持有的 fd。具体来说：
- 子进程保留：本 worker 的任务管道读端、结果管道写端；
- 子进程关闭：本 worker 的任务管道写端、结果管道读端，以及所有其他 worker 的管道 fd。
父进程则相反：
- 父进程保留：本 worker 的任务管道写端、结果管道读端；
- 父进程关闭：本 worker 的任务管道读端、结果管道写端。
这套规则在代码里必须小心落位，因为一旦某个 fd 没有关闭，管道缓冲区和 fd 泄漏问题会在运行一段时间后集中爆发，而且非常难排查。
3. 父进程：任务分发与结果回收的事件循环
3.1 poll 多路复用让事件驱动变得清爽
父进程要同时监听所有 worker 的结果管道，等待任意一个子进程完成当前任务。最笨的办法是依次对每个 result_rfd 做阻塞 read，但这样父进程会被第一个 worker 卡住，其他 worker 完成的结果只能排队等待。另一个笨办法是给每个结果管道设置 O_NONBLOCK，然后不停轮询，这又会让 CPU 空转。
正确的做法是使用 poll 系统调用，把全部 result_rfd 放到一个 pollfd 数组里，一次等待就绪事件：
struct pollfd pfds[WORKER_NUM]; for (int i = 0; i < WORKER_NUM; i++) { pfds[i].fd = workers[i].result_rfd; pfds[i].events = POLLIN; }这样父进程就进入了事件循环：poll 返回后，遍历所有 fd 检查 revents，哪一位被置位就说明对应 worker 完成了任务，立即读取结果、下发新任务。整个父进程不会被任何一个慢 worker 拖住，谁完成了就先处理谁，这是典型的 IO 多路复用思想。
3.2 任务分配逻辑与闲忙标记
任务分配阶段要解决两个问题：任务总数为 TASK_NUM，worker 数为 WORKER_NUM，如何保证每个任务恰好被分发一次，且不会出现重复派发？
我采用的是"初始派发 + 动态补充"的机制：
- 启动后，按顺序给每个 worker 各发一条任务消息，并把这些 worker 标记为 busy；
- poll 等待任意 worker 完成，读到结果后将其标记为空闲；
- 如果还有待发任务（next_task_id < TASK_NUM），立即给这个刚空闲的 worker 发新任务，并再次标记为 busy；
- 如果所有任务都已经派发出去，就只回收结果，不再补充新任务。
这保证了任意时刻，一个 worker 要么空闲没有任何任务，要么正在处理唯一一个任务，绝不存在任务积压在同一 worker 的管道里等待排队。父进程每次 write 任务时，目标 worker 都是空闲的，所以 write 不会因为管道缓冲区满而阻塞。
3.3 优雅关闭：发送退出指令并回收子进程
所有任务处理完之后，进程池要优雅收尾：逐个向子进程下发 CMD_EXIT 退出指令，关闭两端 fd，然后 waitpid 回收子进程。
这里有一个不能省略的细节：父进程必须在 waitpid 之前关闭所有自己持有的任务管道写端。原因很简单，子进程目前在阻塞 read 任务管道，如果父进程不下发退出指令且不关闭写端，子进程永远不会从 read 返回，waitpid 就会永久卡住。先发送 CMD_EXIT 让子进程主动 break，再关闭管道，双保险。
waitpid 本身也值得注意。如果子进程提前处理完任务但没有收到退出指令就结束了，会变成僵尸进程。所以即使子进程崩溃，父进程最后也要 waitpid 回收，避免僵尸进程堆积。
4. 子进程：从管道读取命令到回写结果
4.1 子进程侧代码逻辑
子进程的逻辑比父进程简单得多：一个死循环，阻塞等待任务管道上的消息。拿到任务后执行，然后把结果写入结果管道，继续等待下一条。如果 read 返回 0 表示 EOF（父进程关闭了写端），或者收到 CMD_EXIT 指令，则退出循环。
这里的执行函数可以任意替换。示例里我用平方运算加一点随机 sleep 来模拟耗时任务：
static int do_work(int x) { // 模拟不同耗时的任务 usleep(100000 + (x * 7919) % 200000); return x * x; }子进程在 fork 后要注意设置信号处理：忽略 SIGPIPE。如果父进程提前关闭了结果管道读端（比如父进程崩溃退出），子进程再往结果管道写数据时，内核会发送 SIGPIPE 信号，默认动作是终止进程。忽略这个信号后，write 会返回 -1 并设置 errno 为 EPIPE，子进程可以自行处理退出。
4.2 阻塞读与 EOF 的边界情况
子进程的 read 是阻塞模式，这是进程池设计的核心依赖：没有任务时，子进程不会占用 CPU，只是安静地睡在内核里。这也是 fork 进程池相比忙等轮询的巨大优势。
EOF 的语义需要特别注意。read 返回 0 不仅仅是"没有数据"，而是"对端写端已关闭"。这个信号用于通知子进程"父进程那边已经拆干净了，你可以退出了"。所以子进程的循环退出条件必须同时覆盖三种情况：
- read 返回值小于 0：管道出错，退出；
- read 返回值等于 0：父进程已关闭写端，退出；
- read 返回值大于 0 且消息的 cmd 是 CMD_EXIT：显式退出指令，退出。
只依赖任何单一条件都不够稳健，尤其是只在任务全部分配完后关闭写端、不发 EXIT 消息的做法，虽然也能让子进程退出，但如果子进程还没来得及读空管道，可能错过退出信号。显式消息加 EOF 兜底才是可靠组合。
5. 完整代码与运行效果
5.1 可直接编译的完整源码
下面是一份完整的示例代码，把前面讲的所有设计落实成可运行的程序。任务总数 10 个、worker 数 4 个，每个任务计算参数的平方并返回。
#include <stdio.h> #include <stdlib.h> #include <string.h> #include <unistd.h> #include <signal.h> #include <poll.h> #include <errno.h> #include <sys/wait.h> #define WORKER_NUM 4 #define TASK_NUM 10 #define CMD_EXIT 0 #define CMD_TASK 1 #define CMD_DONE 2 typedef struct { int cmd; int task_id; int arg; } TaskMsg; typedef struct { int task_wfd; int result_rfd; pid_t pid; int busy; } Worker; static int send_task(int task_wfd, int task_id, int arg) { TaskMsg msg; msg.cmd = CMD_TASK; msg.task_id = task_id; msg.arg = arg; ssize_t n = write(task_wfd, &msg, sizeof(msg)); if (n != sizeof(msg)) { perror("write task"); return -1; } return 0; } static void send_exit(int task_wfd) { TaskMsg msg; msg.cmd = CMD_EXIT; msg.task_id = 0; msg.arg = 0; write(task_wfd, &msg, sizeof(msg)); } static int do_work(int x) { usleep(100000 + (x * 7919) % 200000); // 模拟耗时 return x * x; } static void child_main(int task_rfd, int result_wfd) { signal(SIGPIPE, SIG_IGN); TaskMsg msg; while (1) { ssize_t n = read(task_rfd, &msg, sizeof(msg)); if (n <= 0) { break; // EOF 或出错，退出 } if (n != sizeof(msg)) { continue; // 理论上不会发生，防御性处理 } if (msg.cmd == CMD_EXIT) { break; } if (msg.cmd == CMD_TASK) { int result = do_work(msg.arg); TaskMsg reply; reply.cmd = CMD_DONE; reply.task_id = msg.task_id; reply.arg = result; ssize_t w = write(result_wfd, &reply, sizeof(reply)); if (w != sizeof(reply)) { break; } } } close(task_rfd); close(result_wfd); exit(0); } int main() { signal(SIGPIPE, SIG_IGN); Worker workers[WORKER_NUM]; struct pollfd pfds[WORKER_NUM]; for (int i = 0; i < WORKER_NUM; i++) { int task_pipe[2]; int result_pipe[2]; if (pipe(task_pipe) != 0 || pipe(result_pipe) != 0) { perror("pipe"); exit(1); } pid_t pid = fork(); if (pid < 0) { perror("fork"); exit(1); } if (pid == 0) { // 子进程：保留本 worker 的任务读端和结果写端 close(task_pipe[1]); close(result_pipe[0]); child_main(task_pipe[0], result_pipe[1]); // 不会走到这里 } // 父进程：保留本 worker 的任务写端和结果读端 close(task_pipe[0]); close(result_pipe[1]); workers[i].task_wfd = task_pipe[1]; workers[i].result_rfd = result_pipe[0]; workers[i].pid = pid; workers[i].busy = 0; pfds[i].fd = result_pipe[0]; pfds[i].events = POLLIN; } int next_task_id = 0; int task_received = 0; // 初始派发：每个 worker 分一个任务 for (int i = 0; i < WORKER_NUM && next_task_id < TASK_NUM; i++) { if (send_task(workers[i].task_wfd, next_task_id, next_task_id + 1) == 0) { workers[i].busy = 1; next_task_id++; } } while (task_received < TASK_NUM) { int ret = poll(pfds, WORKER_NUM, -1); if (ret < 0) { if (errno == EINTR) { continue; } perror("poll"); break; } for (int i = 0; i < WORKER_NUM; i++) { if (pfds[i].revents & POLLIN) { TaskMsg res; ssize_t n = read(workers[i].result_rfd, &res, sizeof(res)); if (n != sizeof(res)) { fprintf(stderr, "worker %d read result failed: %zd\n", i, n); continue; } printf("worker %d: task %d -> %d\n", i, res.task_id, res.arg); task_received++; workers[i].busy = 0; // 还有任务就补充给刚空闲的 worker if (next_task_id < TASK_NUM) { if (send_task(workers[i].task_wfd, next_task_id, next_task_id + 1) == 0) { workers[i].busy = 1; next_task_id++; } } } else if (pfds[i].revents & (POLLERR | POLLHUP | POLLNVAL)) { fprintf(stderr, "worker %d channel error: revents=0x%x\n", i, pfds[i].revents); } } } // 下发退出指令，关闭 fd，回收子进程 for (int i = 0; i < WORKER_NUM; i++) { send_exit(workers[i].task_wfd); close(workers[i].task_wfd); close(workers[i].result_rfd); } for (int i = 0; i < WORKER_NUM; i++) { waitpid(workers[i].pid, NULL, 0); } printf("all tasks done, worker pool shut down.\n"); return 0; }5.2 编译运行与输出验证
编译命令很简单：
gcc -o pipe_pool pipe_pool.c运行：
./pipe_pool我本地跑一次的结果大致如下：
worker 2: task 2 -> 9 worker 1: task 1 -> 4 worker 3: task 3 -> 16 worker 0: task 0 -> 1 worker 1: task 4 -> 25 worker 3: task 5 -> 36 worker 0: task 6 -> 49 worker 2: task 7 -> 64 worker 2: task 9 -> 100 worker 1: task 8 -> 81 all tasks done, worker pool shut down.可以看到 4 个 worker 都在工作，任务完成顺序是乱序的，因为每个 worker 的 usleep 时间不同。这正是进程池的价值体现：任务被并行处理，而不是父进程一个接一个地同步执行。如果你把 usleep 去掉，改为纯 CPU 计算，还能明显感受到 4 个 worker 并行带来的吞吐提升。
6. 实测中踩过的坑与排查经验
6.1 SIGPIPE 导致进程静默退出
第一次运行这个程序时，任务还没跑完，整个父进程就消失了。检查信号才发现是 SIGPIPE：某次父进程往 worker 的任务管道写数据时，对应 worker 已经因为某种原因退出，写端管道没了读端，内核立刻给父进程发了 SIGPIPE 信号，默认动作是直接终止进程。
这个问题在长时间运行的程序里非常隐蔽，因为进程不是崩溃报错，而是"正常退出"，日志里什么都没有。排查时可以这样验证：
./pipe_pool; echo "exit code: $?"如果 exit code 是 141（128 + 13，13 是 SIGPIPE 的信号编号），就能确认是被 SIGPIPE 杀的。解决办法就是代码里的 signal(SIGPIPE, SIG_IGN)，并检查 write 的返回值。加了这个之后，write 会在对端关闭时返回 -1，errno 是 EPIPE，程序才能从容处理。
6.2 子进程不退出导致的死锁
另一个高频问题是：任务全部跑完后程序卡在 waitpid 不动。我一开始的思路是：所有任务完成后，父进程直接 close 所有 task_wfd，让子进程 read 返回 0 退出，然后 waitpid。听起来没问题，实际却卡住了，因为子进程在创建时继承了所有 worker 的 task_wfd 写端，即使父进程关闭了写端，子进程自己还握着多出来的写端副本，read 永远不会返回 0。
这个问题的根因就是 fd 裁剪没做干净。子进程必须关闭所有不属于自己的写端。如果在 fork 之后没有逐项 close，就一定会踩这个坑。这也是我在代码里强调"fork 后立刻裁剪 fd"的原因，少做一步，换来的就是难以排查的卡死。
排查死锁问题时，用 gdb attach 到进程上，bt 查看调用栈，能看到子进程阻塞在 read 系统调用上，父进程阻塞在 waitpid 上，配合 ps -ef 查看进程树，很快能定位到是管道端点的引用没有被清空。
6.3 poll 返回读写事件的误判
poll 的返回值只告诉我们"有事件发生"，具体是 POLLIN 还是 POLLHUP，要逐个检查 revents。我吃过一次亏：子进程异常退出时，poll 返回的 revents 同时包含 POLLIN 和 POLLHUP，我当时只判断了 POLLIN 就去 read，read 返回 0 也没当成异常处理，导致 task_received 一直达不到 TASK_NUM，程序陷入死循环。
后来我在代码里把 POLLERR、POLLHUP、POLLNVAL 单独列出来处理，一旦读到 EOF 就打印错误日志。这样至少能暴露问题，而不是悄无声息地忙等。这个处理在健壮性上很重要，尤其在真实环境中子进程可能被 kill、被 OOM 杀掉，不会总是按预期退出。
6.4 结果管道冲突与原子写边界
最开始图省事，让所有 worker 共用一根结果管道，结果出现了两个诡异现象：一是父进程读到的结果和 task_id 对应不上，二是偶尔 read 返回的字节数不是 sizeof(TaskMsg)。
根本原因倒不是管道写原子性破了——消息小于 PIPE_BUF 时，单次 write 确实是原子的，两个写者的消息不会在缓冲区里交错占位。但问题在于管道是字节流，单个 read 不保证一次读完"一条"消息。如果两个 worker 几乎同时写入，管道缓冲区里可能是 A 的消息紧跟 B 的消息，父进程 read sizeof(TaskMsg) 字节时取走的可能是 A 整条或者 A 的一部分加上 B 的一部分。
这个例子再次印证了：管道绝不是一个可靠的报文边界协议。要么每次严格读取固定长度的结构体，要么干脆让每个 worker 独占结果管道，从根源上避免多写者共享。我在最终方案里选择了后者，同时配合固定结构体消息，双保险之后就没有再碰到过拼接错乱。
7. 从玩具到可用：这个进程池还能怎么扩展
7.1 把任务从"计算平方"改成任意动作
示例代码里的 do_work 只是简单的平方运算，实际项目中你完全可以把它替换成任何业务逻辑，比如压缩文件、转换图片、解析日志行。一个常见的做法是把任务参数变成一个文件路径字符串，TaskMsg 结构体里放一个足够大的字符数组，子进程收到后打开文件处理。注意要保证 sizeof(TaskMsg) 不要超过 PIPE_BUF，否则就要自己处理消息的分包和重组。
如果你想让任务更通用，可以把 do_work 改成函数指针数组，任务结构体里放一个 function_id，子进程根据 function_id 到函数表中找到对应处理函数，参数统一用 void* 传递。这样进程池就从一个固定业务的小工具，升级成一个通用的任务调度框架。
7.2 动态调整池子大小
固定 worker 数量（WORKER_NUM）在真实场景里不一定合理。你可以用 sysconf(_SC_NPROCESSORS_ONLN) 获取 CPU 核数，再结合任务的 IO 等待比例动态设置。CPU 密集型任务通常设为核数，IO 密集型任务可以设为核数的 2 到 4 倍。
更进一步的方案是实现动态扩容：任务积压时创建新 worker，空闲时回收部分 worker。扩容逻辑就是在父进程端 pipe + fork 一对新管道，加入 poll 监听数组；缩容逻辑则是向目标 worker 发送 CMD_EXIT，关闭管道，waitpid 回收。这套机制与固定池子的核心逻辑完全兼容，改造成本不大。
7.3 与线程池的性能对比与选型建议
用进程池还是线程池，取决于你的核心诉求。线程之间共享地址空间，切换和通信开销更小，但一个线程崩溃可能拖垮整个进程；进程之间完全隔离，某个子进程崩溃不会影响父进程和其他 worker，稳定性更好，代价是占用更多内存、上下文切换成本略高。
我的实际经验是：如果任务是 CPU 密集、彼此独立、生命周期较长，进程池很合适；如果任务需要频繁共享大块数据、追求极致吞吐，线程池可能更好。匿名管道进程池的另一个优势是它不依赖任何第三方库，纯 C 语言加 Linux 系统调用就能实现，非常适合在资源受限的环境里快速落地。
最后再分享一个小技巧：调试进程池时，可以用 strace -f ./pipe_pool 追踪所有子进程的系统调用，看哪一步 read/write 卡住，一目了然。这个工具帮我解决过好几回莫名其妙的"看起来没反应"的问题。
