C语言pipeline如何高效实现进程间通信？

什么是 Pipeline (管道)？

在计算机科学中,Pipeline（管道）是一种进程间通信（IPC, Inter-Process Communication）的机制，它允许一个进程的标准输出 直接连接到另一个进程的标准输入。

你可以把它想象成一个现实中的管道：

• 进程 A 从一端倒入水（输出数据）。
• 进程 B 从另一端接水（输入数据）。
• 水在管道中流动,但两个进程本身并不直接接触。

在命令行中,你肯定见过管道操作符 ，

ls -l | grep ".c"

这条命令的意思是：

1. ls -l 命令执行，列出当前目录的详细信息。
2. ls -l 的输出结果（不再是屏幕显示）被当作 grep ".c" 命令的输入。
3. grep ".c" 从输入中过滤出包含 .c 的行，并显示在屏幕上。

核心思想：将多个简单的、单一功能的命令连接起来，协同完成一个更复杂的任务，在 C 语言中，我们可以使用 pipe() 系统调用来创建这种机制，从而在程序中实现自己的管道。

核心系统调用：pipe()

在 C 语言中，创建管道主要依赖两个系统调用：pipe() 和 fork()。

pipe() 函数

pipe() 函数用于创建一个管道，它会在内核中开辟一块缓冲区，用于数据传输。

函数原型：

#include <unistd.h>
int pipe(int pipefd[2]);

参数：

• pipefd: 一个整型数组，由调用者提供，函数执行成功后，pipefd[0] 将代表管道的读端，pipefd[1] 将代表管道的写端。

返回值：

• 成功：返回 0。
• 失败：返回 -1，并设置 errno。

重要规则：

1. 数据流向：数据只能从 pipefd[1]（写端）写入，从 pipefd[0]（读端）读出。
2. 单向性：管道是半双工的，如果你想实现双向通信，需要创建两个管道。
3. 缓冲区大小：管道有一个固定大小的内核缓冲区（通常为 64KB），如果写得太快而读得太慢，写操作可能会被阻塞，直到有空间为止，反之，如果读得太快而写得太慢，读操作可能会读到文件结束符（EOF）。

一个经典的 Pipeline 模式：父进程写，子进程读

这是最基础的管道使用场景,通常用于父进程向子进程传递数据。

工作流程：

1. 父进程调用 pipe() 创建管道，得到 fd[0]（读）和 fd[1]（写）。
2. 父进程调用 fork() 创建一个子进程。
3. 子进程继承父进程的文件描述符表，父子进程都拥有对 fd[0] 和 fd[1] 的访问权。
4. 关键步骤：子进程关闭写端 fd[1]，因为它只需要从管道中读取数据，父进程关闭读端 fd[0]，因为它只需要向管道中写入数据。这一步至关重要，可以避免死锁和资源浪费。
5. 父进程通过 write(fd[1], ...) 向管道写入数据。
6. 子进程通过 read(fd[0], ...) 从管道中读取数据。
7. 通信完成后,双方都关闭各自剩下的文件描述符。

代码示例：父进程写，子进程读

下面是一个完整的 C 语言示例，实现了上述流程。

#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>
#include <sys/wait.h>
#define BUFFER_SIZE 256
int main() {
    int pipefd[2]; // 用于存放管道的文件描述符
    pid_t pid;
    char write_buf[] = "Hello from parent process!";
    char read_buf[BUFFER_SIZE];
    // 1. 创建管道
    if (pipe(pipefd) == -1) {
        perror("pipe");
        exit(EXIT_FAILURE);
    }
    // 2. 创建子进程
    pid = fork();
    if (pid == -1) {
        perror("fork");
        exit(EXIT_FAILURE);
    }
    if (pid > 0) {
        // 父进程
        printf("Parent process (PID: %d) is running.\n", getpid());
        // 3. 父进程关闭读端
        close(pipefd[0]);
        // 4. 父进程向管道写入数据
        write(pipefd[1], write_buf, strlen(write_buf));
        printf("Parent: Sent message to child: \"%s\"\n", write_buf);
        // 5. 关闭写端，表示数据发送完毕
        close(pipefd[1]);
        // 等待子进程结束
        wait(NULL);
    } else if (pid == 0) {
        // 子进程
        printf("Child process (PID: %d) is running.\n", getpid());
        // 3. 子进程关闭写端
        close(pipefd[1]);
        // 4. 子进程从管道读取数据
        int nbytes = read(pipefd[0], read_buf, BUFFER_SIZE);
        if (nbytes > 0) {
            read_buf[nbytes] = '\0'; // 确保字符串正确终止
            printf("Child: Received message from parent: \"%s\"\n", read_buf);
        }
        // 5. 关闭读端
        close(pipefd[0]);
        exit(EXIT_SUCCESS);
    }
    return 0;
}

编译和运行：

gcc parent_child_pipe.c -o pipe_demo
./pipe_demo

预期输出：

Parent process (PID: 1234) is running.
Child process (PID: 1235) is running.
Parent: Sent message to child: "Hello from parent process!"
Child: Received message from parent: "Hello from parent process!"

(注意：PID 号会每次运行都不同)

进阶：子进程写，父进程读

这和上面的模式是镜像的,只需要在 fork() 之后，让子进程关闭读端，父进程关闭写端即可。

代码逻辑调整：

• 在 if (pid > 0) (父进程) 中：close(pipefd[1]); read(pipefd[0], ...);
• 在 else if (pid == 0) (子进程) 中：close(pipefd[0]); write(pipefd[1], ...);

更复杂的 Pipeline：命令 A | B | C

在命令行中,我们可以串联多个管道，A | B | C，这意味着：

• A 的输出 -> B 的输入
• B 的输出 -> C 的输入

在 C 语言中实现这个，我们需要创建两个管道，并管理多个子进程。

工作流程：

1. 创建管道 pipe1 和管道 pipe2。
2. 创建子进程 pid1 (执行命令 B)。
3. 在 pid1 中：
   • dup2(pipe1[0], STDIN_FILENO); // 将 pipe1 的读端重定向到标准输入
   • dup2(pipe2[1], STDOUT_FILENO); // 将 pipe2 的写端重定向到标准输出
   • 关闭所有不必要的文件描述符。
   • 执行 exec() 来运行命令 B。
4. 创建子进程 pid2 (执行命令 C)。
5. 在 pid2 中：
   • dup2(pipe2[0], STDIN_FILENO); // 将 pipe2 的读端重定向到标准输入
   • 关闭所有不必要的文件描述符。
   • 执行 exec() 来运行命令 C。
6. 在父进程中：
   • dup2(pipe1[1], STDOUT_FILENO); // 将标准输出重定向到 pipe1 的写端
   • 关闭所有不必要的文件描述符。
   • 执行 exec() 来运行命令 A。
7. 父进程等待所有子进程结束。

关键点：dup2() dup2(oldfd, newfd) 是一个强大的系统调用，它会用 oldfd 来复制 newfd。newfd 已经打开，它会被先关闭，这使得我们将管道的端口重定向到标准输入 (STDIN_FILENO, 0) 或标准输出 (STDOUT_FILENO, 1) 变得非常方便。

总结与注意事项

常见问题：

1. 死锁：最常见的错误是忘记关闭不需要的管道端口，如果父子进程都保留了写端，并且都试图写入，或者都等待对方读取，就会导致死锁。
2. read() 阻塞：如果管道的写端都已关闭，read() 会立即返回 0（表示 EOF），如果写端没有关闭且没有数据，read() 会阻塞。
3. write() 阻塞：如果管道的缓冲区已满，write() 会阻塞，直到有空间被读走。
4. 管道的局限性：
   • 它是半双工的。
   • 它只能在具有亲缘关系的进程间使用（通常是父子进程或兄弟进程）。
   • 它的生命周期随进程,如果所有使用管道的进程都结束了，管道也就被销毁了。

掌握了 pipe() 和 fork() 的配合使用，你就可以在 C 语言中创建强大的、类似 shell 命令行的数据处理流程。