4、管道实战进阶：双管道实现全双工通信、管道与select/poll结合、管道与epoll结合、常见管道bug排查
管道这东西，基础用法大家都会。但真正到了实战中，你会发现单管道那点单向通信能力，根本不够用。我刚开始做多进程开发时，就吃过这个亏——两个进程互相发消息，结果死锁了，查了半天才发现是管道方向搞反了。
这一章，咱们就把管道玩透。从双管道实现全双工，到和I/O多路复用结合，再到那些让人头疼的bug排查。嗯，都是我在项目中踩过的坑。
4.1 双管道实现全双工通信
普通管道是单向的。你想让两个进程互相发消息？那就得开两条管道。说白了，一条从A到B，一条从B到A。
核心思路：父进程创建两个管道，一个用于父写子读，一个用于子写父读。这样双方都能收发数据。
我给大家看个实际例子。这个代码我当年在写一个日志采集系统时用过，两个进程互相传递控制命令和数据流：
#include <stdio.h>
#include <unistd.h>
#include <sys/wait.h>
#include <string.h>
int main() {
int pipe1[2]; // 父写 -> 子读
int pipe2[2]; // 子写 -> 父读
pid_t pid;
char buf[256];
if (pipe(pipe1) == -1 || pipe(pipe2) == -1) {
perror("pipe");
return 1;
}
pid = fork();
if (pid == 0) {
// 子进程
close(pipe1[1]); // 关闭pipe1写端
close(pipe2[0]); // 关闭pipe2读端
// 从pipe1读，往pipe2写
read(pipe1[0], buf, sizeof(buf));
printf("子进程收到: %s\n", buf);
strcpy(buf, "收到，over");
write(pipe2[1], buf, strlen(buf) + 1);
close(pipe1[0]);
close(pipe2[1]);
} else {
// 父进程
close(pipe1[0]); // 关闭pipe1读端
close(pipe2[1]); // 关闭pipe2写端
strcpy(buf, "开始干活");
write(pipe1[1], buf, strlen(buf) + 1);
read(pipe2[0], buf, sizeof(buf));
printf("父进程收到: %s\n", buf);
close(pipe1[1]);
close(pipe2[0]);
wait(NULL);
}
return 0;
}个人经验：我习惯在fork之后立刻关闭不需要的管道端。这样做有两个好处：一是节省文件描述符，二是避免误操作。你想想看，如果子进程还留着写端没关，父进程读数据时可能永远等不到EOF。
4.2 管道与select/poll结合
单管道读写是阻塞的。如果管道里没数据，read会一直卡着。这在多进程通信中很要命——你总不能专门开个线程去等数据吧？
这时候就需要select或poll出场了。它们能帮你同时监控多个管道，哪个有数据就读哪个。我在做一个网关程序时，同时监控了8个管道，全靠select撑起来的。
4.2.1 使用select监控管道
#include <stdio.h>
#include <unistd.h>
#include <sys/select.h>
#include <string.h>
int main() {
int pipe_fd[2];
fd_set read_fds;
struct timeval tv;
char buf[256];
pipe(pipe_fd);
pid_t pid = fork();
if (pid == 0) {
// 子进程：写数据
close(pipe_fd[0]);
sleep(2); // 模拟延迟
write(pipe_fd[1], "hello", 6);
close(pipe_fd[1]);
} else {
// 父进程：用select等数据
close(pipe_fd[1]);
FD_ZERO(&read_fds);
FD_SET(pipe_fd[0], &read_fds);
tv.tv_sec = 5;
tv.tv_usec = 0;
int ret = select(pipe_fd[0] + 1, &read_fds, NULL, NULL, &tv);
if (ret == -1) {
perror("select");
} else if (ret == 0) {
printf("超时了，没等到数据\n");
} else {
if (FD_ISSET(pipe_fd[0], &read_fds)) {
read(pipe_fd[0], buf, sizeof(buf));
printf("收到数据: %s\n", buf);
}
}
close(pipe_fd[0]);
wait(NULL);
}
return 0;
}避坑指南：我曾经犯过一个低级错误——select之后忘记重新设置timeout。select会修改timeval结构体，把剩余时间写进去。如果你在循环里用select，每次都要重新初始化timeout。不然第二次调用时timeout可能已经是0了，直接超时返回。
4.2.2 使用poll监控管道
poll比select好一点的地方在于，它没有文件描述符数量限制。select默认只能监控1024个，poll就没这个问题。我个人现在写新代码基本都用poll了。
#include <stdio.h>
#include <unistd.h>
#include <poll.h>
#include <string.h>
int main() {
int pipe_fd[2];
struct pollfd fds[1];
char buf[256];
pipe(pipe_fd);
pid_t pid = fork();
if (pid == 0) {
close(pipe_fd[0]);
sleep(1);
write(pipe_fd[1], "world", 6);
close(pipe_fd[1]);
} else {
close(pipe_fd[1]);
fds[0].fd = pipe_fd[0];
fds[0].events = POLLIN;
int ret = poll(fds, 1, 3000); // 3秒超时
if (ret == -1) {
perror("poll");
} else if (ret == 0) {
printf("超时\n");
} else {
if (fds[0].revents & POLLIN) {
read(pipe_fd[0], buf, sizeof(buf));
printf("收到: %s\n", buf);
}
}
close(pipe_fd[0]);
wait(NULL);
}
return 0;
}4.3 管道与epoll结合
如果你要监控的管道数量特别多，比如几百上千个，那select和poll都不够看了。这时候就得请出epoll。epoll是Linux下最高效的I/O多路复用机制，没有之一。
我记得有一次做一个消息中间件，要同时管理500多个子进程的管道通信。用poll的时候CPU占用率居高不下，换成epoll之后，直接降到了个位数。差距就是这么大。
#include <stdio.h>
#include <unistd.h>
#include <sys/epoll.h>
#include <string.h>
#define MAX_EVENTS 10
int main() {
int pipe_fd[2];
int epoll_fd;
struct epoll_event ev, events[MAX_EVENTS];
char buf[256];
pipe(pipe_fd);
pid_t pid = fork();
if (pid == 0) {
close(pipe_fd[0]);
sleep(1);
write(pipe_fd[1], "epoll test", 11);
close(pipe_fd[1]);
} else {
close(pipe_fd[1]);
// 创建epoll实例
epoll_fd = epoll_create1(0);
if (epoll_fd == -1) {
perror("epoll_create1");
return 1;
}
ev.events = EPOLLIN;
ev.data.fd = pipe_fd[0];
if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, pipe_fd[0], &ev) == -1) {
perror("epoll_ctl");
return 1;
}
// 等待事件，超时5秒
int nfds = epoll_wait(epoll_fd, events, MAX_EVENTS, 5000);
if (nfds == -1) {
perror("epoll_wait");
} else if (nfds == 0) {
printf("超时\n");
} else {
for (int i = 0; i < nfds; i++) {
if (events[i].events & EPOLLIN) {
read(events[i].data.fd, buf, sizeof(buf));
printf("epoll收到: %s\n", buf);
}
}
}
close(epoll_fd);
close(pipe_fd[0]);
wait(NULL);
}
return 0;
}使用建议：epoll有两种触发模式——水平触发（LT）和边缘触发（ET）。默认是LT，和select/poll行为一致。ET模式效率更高，但使用起来要小心：你必须一次性把数据读完，否则就再也收不到事件通知了。我建议新手先用LT，等熟悉了再尝试ET。
4.4 常见管道bug排查
管道用多了，bug也就见多了。我把自己这些年踩过的坑整理一下，希望能帮你少走弯路。
4.4.1 死锁问题
最常见的就是死锁。两个进程互相等对方写数据，结果谁都不写。为什么会这样？说白了就是管道满了。
Linux管道默认缓冲区是65536字节（64KB）。如果写端一直写，读端不读，缓冲区满了之后write就会阻塞。这时候如果读端也在等写端发消息，那就死锁了。
解决方案：要么用非阻塞I/O，要么用select/poll/epoll做超时控制。我个人的习惯是，所有管道读写都加上超时，宁可超时重试，也不要死锁。
4.4.2 文件描述符泄漏
这个bug特别隐蔽。你fork之后，子进程继承了父进程的所有文件描述符。如果子进程不关掉不需要的管道端，这些fd就一直占着。时间长了，系统fd用完了，pipe()就返回-1。
我曾经在一个长期运行的服务里遇到这个问题。跑了三天，突然所有管道都创建失败了。查了半天，发现是某个子进程没关父进程的管道写端。
4.4.3 数据粘包问题
管道是字节流，没有消息边界。如果你连续写两次，读端可能一次就读走了。比如你写了"hello"和"world"，读端可能读到"helloworld"。
解决办法有几种：
- 固定消息长度：每次读写固定字节数
- 添加消息头：头部包含消息长度信息
- 使用特殊分隔符：比如用'\n'作为消息结束标志
我个人推荐第二种方式，加消息头。虽然实现起来复杂一点，但最可靠。
4.4.4 SIGPIPE信号问题
如果读端关闭了管道，写端还在写，系统会发送SIGPIPE信号。默认行为是终止进程。很多新手第一次遇到这个问题时都懵了——进程莫名其妙就挂了。
避坑指南：如果你不想让进程被SIGPIPE干掉，可以忽略这个信号：signal(SIGPIPE, SIG_IGN);。这样write会返回-1，errno设置为EPIPE。你就可以自己处理这个错误了。
4.4.5 缓冲区溢出
管道缓冲区有限，写太快读太慢就会满。这时候write会阻塞。如果你用的是非阻塞模式，write会返回-1，errno是EAGAIN。
我建议的做法是：写数据前先检查管道是否可写。用select或poll监控写事件，等管道有空位了再写。这样既不会阻塞，也不会丢数据。
| 常见bug | 症状 | 解决方案 |
|---|---|---|
| 死锁 | 进程卡住不动 | 加超时、用非阻塞I/O |
| fd泄漏 | pipe()返回-1 | 及时关闭不需要的管道端 |
| 数据粘包 | 数据混在一起 | 加消息头或分隔符 |
| SIGPIPE | 进程意外退出 | 忽略SIGPIPE信号 |
| 缓冲区溢出 | write阻塞或返回EAGAIN | 监控写事件，按需写入 |
嗯，管道这东西，说简单也简单，说复杂也复杂。关键是要理解它的特性——单向、字节流、有缓冲区大小限制。把这些搞明白了，再用select/poll/epoll武装起来，基本上就能应对大部分场景了。
下一章我们会讲命名管道（FIFO），那又是另一个世界了。先消化好今天的内容，尤其是双管道和epoll的用法，后面会经常用到。