使用poll()实现多子进程管道读写及输出交错的C语言示例

核心逻辑拆解 #

原来的串行输出问题本质是：每个子进程独立运行完整个命令文件后，父进程按文件顺序逐个读取管道内容，天然会把单个文件的所有输出一次性全部读完，完全打乱行级的交叉顺序。要实现严格的A→C→B→D输出顺序，不需要让子进程每跑一行就停下来同步，只要父进程侧做行级的缓存调度即可，整体流程如下：

• 预解析所有传入的命令文件，逐行统计每个文件的命令总行数，建立基础的行索引映射。
• 为每个文件创建独立管道，fork出专属子进程：子进程关闭管道读端，把stdout/stderr重定向到管道写端，通过exec启动bash执行对应命令文件，完全并行运行，不设置任何中间阻塞。
• 父进程关闭所有管道的写端，把所有管道读端加入poll监控列表，不绑定固定的读取顺序。
• 父进程为每个读端维护独立的字节缓存，用poll异步等待任意管道的可读事件，每次读到数据就按换行符切分成独立行存在缓存里。
• 父进程维护全局行游标，按「文件1第N行→文件2第N行→…→文件M第N行」的固定顺序，检查每个文件的缓存里有没有准备好当前游标对应的行，准备好就直接打印，游标同步往后推进。
• 直到所有管道都收到POLLHUP事件（所有子进程执行完毕）、所有缓存内容都按顺序输出完成，程序退出。

poll()配合管道的关键使用细节 #

很多人用poll踩坑都是没搞清楚事件和fd管理的逻辑，核心要注意这几点：

• 每个管道的读端对应一个独立的pollfd结构，events字段直接设为POLLIN | POLLHUP即可，不需要额外配置：前者代表有新数据可读，后者代表对端（子进程侧的管道写端）已经全部关闭，不会再有新数据发来。
• 调用poll()时超时参数设为-1，无限等待任意事件触发即可，不需要手动算超时，也不需要像select那样每次调用都重置监控集合。
• 事件返回后要逐条目检查revents字段：
  • 收到POLLIN时，单次最多读4KB数据追加到对应缓存即可，不要一次性读到阻塞，避免卡住其他管道的处理。
  • 收到POLLHUP时，把这个条目的fd字段设为-1，后续poll就会自动跳过这个条目，等所有剩余缓存读完就不会再触发事件。
  • 收到POLLERR时直接关闭对应fd，打印错误信息就行，不需要额外处理。
• 用完的事件一定要把revents字段清零，不然下次poll调用会误判事件状态。

核心实现代码片段 #

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <poll.h>
#include <string.h>
#include <sys/wait.h>

// 单个命令文件对应的上下文
typedef struct {
    int pipe_read_fd;    // 管道读端
    const char* file_path;
    char* line_cache;    // 未输出的行缓存
    size_t cache_len;    // 缓存当前长度
    int total_lines;     // 文件里的总命令行数
} FileContext;

int main(int argc, char* argv[]) {
    // 省略：参数校验、文件解析、建管道、fork子进程的前置逻辑
    int file_cnt = argc -1;
    FileContext* ctx = calloc(file_cnt, sizeof(FileContext));
    struct pollfd* pfds = calloc(file_cnt, sizeof(struct pollfd));

    // 初始化poll监控列表
    for (int i = 0; i < file_cnt; i++) {
        pfds[i].fd = ctx[i].pipe_read_fd;
        pfds[i].events = POLLIN | POLLHUP;
    }

    int row_cursor = 0;
    int all_done = 0;

    while (!all_done) {
        // 等待事件触发，无限阻塞
        int ret = poll(pfds, file_cnt, -1);
        if (ret == -1) {
            perror("poll调用失败");
            break;
        }

        // 处理所有触发的事件
        for (int i = 0; i < file_cnt; i++) {
            if (pfds[i].revents == 0) continue;
            FileContext* cur = &ctx[i];

            // 处理可读数据
            if (pfds[i].revents & POLLIN) {
                char tmp[4096];
                ssize_t read_n = read(pfds[i].fd, tmp, sizeof(tmp)-1);
                if (read_n > 0) {
                    // 动态扩容缓存，追加读到的内容
                    cur->line_cache = realloc(cur->line_cache, cur->cache_len + read_n +1);
                    memcpy(cur->line_cache + cur->cache_len, tmp, read_n);
                    cur->cache_len += read_n;
                    cur->line_cache[cur->cache_len] = '\0';
                }
            }

            // 对端关闭，标记为从监控中移除
            if (pfds[i].revents & POLLHUP) {
                pfds[i].fd = -1;
            }
            pfds[i].revents = 0; // 清空事件标记
        }

        // 按顺序输出逻辑
        int has_output = 1;
        while (has_output) {
            has_output = 0;
            // 同一行游标下，按文件顺序依次检查缓存
            for (int i =0; i < file_cnt; i++) {
                FileContext* cur = &ctx[i];
                if (row_cursor >= cur->total_lines) continue;
                // 检查缓存里有没有完整的一行
                char* line_end = strchr(cur->line_cache, '\n');
                if (line_end) {
                    *line_end = '\0';
                    printf("%s\n", cur->line_cache);
                    // 把已输出的内容从缓存移除
                    size_t remain_len = cur->cache_len - (line_end +1 - cur->line_cache);
                    memmove(cur->line_cache, line_end +1, remain_len);
                    cur->cache_len = remain_len;
                    cur->line_cache[cur->cache_len] = '\0';
                    has_output = 1;
                }
            }
            // 本轮所有同游标的行都输出完，游标推进
            if (has_output) {
                row_cursor++;
            }
        }

        // 检查是不是所有监控的fd都已经关闭
        int all_closed =1;
        for (int i=0; i<file_cnt; i++) {
            if (pfds[i].fd != -1) {
                all_closed =0;
                break;
            }
        }
        all_done = all_closed;
    }

    // 回收所有子进程，避免僵尸进程
    while (wait(NULL) >0);

    // 省略：资源释放逻辑
    free(ctx);
    free(pfds);
    return 0;
}

高频踩坑提醒 #

• 子进程重定向stdout到管道写端时，一定要记得关闭管道读端，同时给写端设置FD_CLOEXEC标记，否则exec之后子进程会持有多余的写端引用，父进程永远等不到POLLHUP事件，程序直接卡死。
• 行缓存一定要用动态扩容的逻辑，不要用固定大小的栈数组，一旦命令输出长度超过数组大小就会内存越界。
• 输出顺序不能改成从单个缓存里读第一行就打印，那样又会回到“一个文件的输出全部打完再打另一个”的老问题，必须严格按“同游标的文件顺序”去检查缓存。
• 最后一定要循环调用wait回收所有子进程，不然会留下一堆僵尸进程。

内容的提问来源于stack exchange，提问作者JavaDumbell