当前位置：首页 > news >正文

【Linux笔记】进程间通信之管道

news 2025/11/11 1:24:53

一、匿名管道

我们在之前学习进程的时候就知道了一个概念，就是进程间是互相独立的，所以就算是两个进程是父子关系，其中一个进程退出了也不会影响另一个进程。

也因为进程间是互相独立的，所以两个进程间就不能直接的传递信息或者互相访问各自的资源，而如果想要让两个进程之间进行信息传递，就需要用到我们今天要讲的“进程间通信”。

实现进程间通信其实有很多种方法，我们今天讲的“管道”其实是一种早期的进程间通信方式，但是对于我们学习后面的进程间通信具有指导作用。

1.1、进程间通信的本质

进程间不能直接通信，但是我们知道操作系统是进程的管理者，如果我们想要让两个进程传递信息，我们可以让操作系统作为“中间人”啊。

所以一个进程想要和另一个进程传递信息，可以先将要传递的信息交给操作系统，再由操作系统将信息传递给另一个进程。

那么这个资源就必定会让两个进程都能访问到，所以进程间通信的本质是：让两个进程看到同一份资源！

而这份资源，通常是由操作系统提供的，而我们进程要讲的“管道”其实就是这一份资源，管道其实是一个文件。

1.2、匿名管道的原理

上面说到，管道其实是一个文件，那么想要通过匿名管道进程进程间通信就一定是要让两个进程同时打开这个匿名管道文件，所以一个进程想要个另一个进程进行信息传递，就可以先将信息写入到匿名管道文件中，另一个进程则是从这个匿名管道文件中读取。

而我们也知道，操作系统会为每一个被打开的文件创建一个文件缓冲区，所有向该文件中写入的内容都不是直接写入磁盘中，而是先存储在文件缓冲区中，所以匿名管道文件实际上只是通过缓冲区来实现信息的传递。事实上匿名管道文件也只能通过缓冲区来实现信息传递，而且匿名管道文件也并不需要在磁盘中存储。

此外，匿名管道文件还有一些特点：

1、匿名管道文件是可以被同一个进程打开两次的文件

匿名管道文件打开两次其实并不是同一个匿名管道文件在内存中被加载两次，而是同一个匿名管道文件会有两个struct file结构体对象，一个用来读、一个用来写，分别是读端和写端。

如下图所示：

2、匿名管道只能完成具有亲缘关系的进程间的通信

这个特性其实是由匿名管道的“设计”而决定的，因为“匿名”管道其实也就是没有名字也没有路径的管道文件，因为这个文件没有名字也没有路径，这也就注定了其他进程不可能通过名字和路径找到该文件，再结合我们之前学过的文件描述符，我们也只能通过父进程创建子进程，然后子进程继承父进程的文件描述符表的方式让其他进程(子进程)拿到该文件的文件描述符，也就相当于找到了这个文件。

如下图所示：

2、匿名管道文件只能实现单向通信

其实无论是匿名管道或命名管道都是只支持单向通信的，这样设计是为了更方便。

所以一旦确定了数据传输的方向，为了保险，我们就需要将父子进程多余的不需要的文件描述符关闭，比如我们进程想要让父进程向管道里写，让子进程从管道中读。那我们就需要将父进程的读端关闭，将子进程的写端关闭：

所以这也就是为什么一个匿名管道文件被打开的时候会将读端和写端都打开，因为只有这样我们才能让子进程也继承到读端和写端，后面也就能选择性的关闭相应的文件描述符从而形成特定方向的信道了。

1.3、匿名管道的使用与特性

想要在我们的程序中打开一个匿名管道文件，我们首先要认识一个系统调用接口：

这个pipe会为我们打开一个匿名管道文件并返回读端和写端的文件描述符，只不过返回的文件描述符我们需要使用一个长度为2的文件描述符数组接收，也就是上面的pipefd[2]，这个pipefd其实是一个输出型参数，并且规定，pipdfd[0]存储的是读端的文件描述符，pipefd[1]存储的是写端的文件描述符。

所以，有了上面这个接口和前面的理论，我们就可以在我们的程序中创建一个单向通讯的信道了，今天我们规定这个信道的方向是父进程读取，子进程写入：

至此，我们就创建好了一个单向通信的信道，然后就可以进行进程间通信了。

我们先写一个简单的例子，让子进程一直向父进程发送动态的数据，并且让父进程读取并打印出来：

我们先运行一下试试：

但是运行后我们会发现打印出来的信息非常的乱，这是因为我们现在的写端写的太快了，一下自救把信息给写完了，所以读端一读就一下子将所有的信息都读了出来。

所以我们可以让写端写的慢一点，让它写完后暂停一秒：

我们再次运行：

我们看到这次的运行就打印得好多了，但是后面确卡死了。

我们先不想为什么会卡死，大家难道没发现一个问题吗？

我们上面只是让子进程在写的时候暂停了，而并没有让父进程也暂停，按理说父进程也在一直打印啊，但为什么父进程也只打印了10次呢？

这是因为操作系统会默认给写端提供“同步”机制，也就是说当管道中没有数据时，读端必须等待，直到管道中有数据为止。

上面的读端虽然一直在读，但是我们现在的数据量很小，读端一读就将管道中所有的数据全都读完了，读完管道也就“清空”了，再次读的时候写端还没向管道中写数据，所以读端就只能等待。

注意是只有写端哦，也就是说只有读端会等待写端，而写端不会等待读端，就比如我们接下来让写端写的很快，而读端读得很慢：

那么结果就会打印的非常乱了：

这就说明了，写端并不会等待读端，虽然你读端并没有在读，但我写端却一直在写。

那最后为什么又会卡住呢？

这是因为，正常情况下如果写端关闭，读端一直在读，read就会返回0，表示读到文件结尾。

而我们现在没有多返回值为0做判断，所以也就卡住了，我们多加一个判断即可：

这样进程就能正常退出了：

上面是写端关闭的情况，那我们让读端中途关闭又会发生什么呢？

我们干脆就只让读端只读一次，就直接关闭：

我们会发现读端在读了一次之后整个进程就直接退出了：

这样不好看出是为什么会出退，我们可以让父进程在关闭读端后暂停一会，然后通过监控脚本来看看子进程的情况：

从结果从结果中我们可以看到，子进程在父进程关闭读端后直接就崩溃了。

这是因为，读端关闭，写端一直写。操作系统会直接杀掉写端进程，方法是通过向写端发送SIGPIPE(13)信号终止进程。

那么如何证明呢？

很简单，就是用我们以前学过的进程等待接口waitpid的第二个参数接收进程退出的退出信号：

以上就是匿名管道的几个重要的特性，讲完了这些之后我们再来理解一下我们操作系统指令中的一个‘|’匿名管道：

‘|’管道相信大家都用过，但是可能都不知道原理，其实它的原理其实就是我们上面所说到的匿名管道的原理，只是多加了一个重定向。

上面的这些指令本质其实都是创建了几个进程，然后通过管道进行进程间通信，只不过这几个进程都是命令行解释器bash创建的，上面的例子不好看出来，因为是一执行就退出的，使用下面这个例子就会看的清楚一点：

同时，bash在创建这进程和管道的同时，还对这几个进程做了输入输出重定向，就拿比较简单的echo | cat来说，其结构如下图所示：

bash在底层其实是将echo的标准输出和管道的写端进行交换，所以此后进程echo所执行的一切输出到显示器中的信息就都会写入到管道文件中，而在另一端则是将进程cat的标准输入和管道的读端进行交换，所以此后进程cat执行的一切从标准输入中读取数据的工作就都会从管道文件中读取。

所以我们才能看到原本要由echo输出到显示器上的内容反而是被cat打印到了显示器上的：

所以我们也就能知道了为什么grep指令能从管道前面执行的结果中过滤出相关信息：

二、使用匿名管道实现一个简易的“进程池”

既然我们已经了解了匿名管道的原理，那当然也要来做一些实践，只有原理和实践结合起来，我们才能加深对匿名管道的理解。

其实管道的应用中有一个非常好的例子，那就是实现一个“进程池”。

2.1、什么是进程池？

大家在学习C++的时候可能听说过空间配置器，也可能听说过进程池，但可能并不真正理解这些到底是个什么东西。

通俗的理解，我们可以将进程池理解成一个内存储备库。

我们在学习C语言的时候用过malloc，学习C++的时候用过new，它们两个的作用都是申请一块内存空间供用户使用。

而且我们也知道操作系统是内存的管理者，所以分配内存资源的工作只能有操作系统来做，而我们要访问操作系统只能通过系统调用。

那么如果我们要频繁使用malloc或new来申请空间，难道是每次都要调用系统调用吗？

其实用我们目前的知识来分析，可以推断出如果是这样做，效率其实是不太高的，因为调用系统调用也是有消耗的，可能单次的影响不是很大，但是多次的话就会有一点影响了。

那再结合我们之前学过的缓冲区的知识，我们的内存是否也可以像缓冲区一样，先申请好一部分资源，到用到的时候再分配。

其实是可以的，事实上进程池的原理和缓冲区确实很类似，比如说我们今天有很多的任务要执行，要将这些任务分配给不同的进程去执行，那我们就可以事先先创建好一批进程，然后从这里进程中选择一些进程执行某些任务，没有收到任务的进程就一直等待任务的到来。这样岂不是比需要一个进程再创建一个进程好多了。

而这就是进程池。

那我们今天要怎样实现一个进程池呢？我们可以先将图画出来：

如上图，进程池的创建其实都是由父进程来完成的，让父进程一次创建一个管道，在创建完一个管道之后立即创建一个子进程，然后做好相应的链接工作。

而我们往后要做的是让父进程给子进程派发任务，那当然是父进程往管道里写信息，然后让子进程来读了，所以对于每一个新创建的管道，父进程要关闭它的读端，子进程要关闭它的读端。

2.2、进程池的模拟实现

2.2.1、创建管道创建进程形成单向信道

第一步我们先要吧管道和子进程先创建出来，经过上面的描述，完成这一步也不是什么难事：

2.2.2、管理管道管理任务

我们上面已经好了管道和进程并形成了单向通信的信道，而我们的目标是让父进程可以选择子进程发送任务并让子进程执行，那我们就必须将这些管道和子进程管理起来，这样将来也好让父进程找到。

所以我们可以设计一个结构体channel，在这个结构体中记录一个管道和进程的相关信息，然后我们每创建一个管道和子进程就创建一个channel结构体对象。最后再将这些结构体对象统一管理起来，比如放到一个数组里面。

我们先来创建这样一个channel结构体类型：

可能有的朋友会不知道上面结构体中这个ctrfd具体应该存的是什么，那么我在这里做一些解释。

这是因为父进程是需要创建多个管道的，而每创建一个管道都会占用父进程的文件描述符表，所以之前的图如果画得更详细一点，应该是这样的：

而父进程往后是要向管道中写入数据，那么那么父进程就需要找到每个管道对应的写端文件描述符，所以channel结构体中的ctrfd存的就是该管道的写端文件描述符(在父进程中)。

所以我们就可以创建一个管道数组，在每次父进程创建完子进程之后就向该数组中加入一个管道对象：

完成之后，我们可以试着将这个数组打印出来检查：

那下一步就是要创建出一些”任务“来让子进程执行，对于”任务”呢我们今天就实现得简单一些，让子进程输出一些信息就行了，例如：

不要搞得太复杂，意思意思就行了。

而这些任务将来也是需要被父进程找到的，所以我们可以顶一个包装器，定义一个包装器数组，然后将这些任务全都放到这个数组中，这样我们以后就可以直接使用数组下标来选择对应的任务了：

2.2.3、分配任务

再分配任务给子进程之前，我们必须先让子进程一直等待任务的到来，而我们之前在说管道的特性的时候说过，如果管道中没有数据，那么读端就必须等待，直到管道中有数据为止，所以我们就让子进程一直从管道中读取就行了。

而父进程最终一定是要先选择某一个管道，然后再向管道中发送信息，让对应的子进程执行某一个任务，而我们刚才又已经将任务以数组的方式组织起来了，所以子进程需要读的就只是某一个任务对应的下标即可：

而如果，再读的过程中写端关闭了，根据之前总结的特性，读端会读到0，那既然写端都关闭了，那我们就不需要再读了，直接让子进程跳出循环然后退出即可。

而我们父进程需要做的就是选择管道，选择任务，最后只需要向对应的管道中发送任务的下标即可。

选择的方式有很多种，我们可以随机的选择，也可以以轮询的方式选择，我们今天以为轮询的方式选择管道，然后以随机的方式选择任务：

运行结果：

当然，我们差最后一步，那就是在进程退出之前，将所有的管道关闭和所有的子进程回收，而我们已经将进程和管道的信息都保存在了，channels数组中了，所以只需要遍历一遍channels数组即可：

运行结果：

而上面我为什么要遍历两次channels数组，而不是只遍历了一遍，以及为什么需要先关闭管道再回收进程，这就得好好说一下了，因为这里面还藏着一个隐藏的非常深的bug。

2.2.4、处理一个比较“深”的bug

如果我们现在先回收子进程，再关闭管道，我们就能看到这个bug了：

我们会发现最后程序卡住了：

同样的，如果我们将回收子进程和关闭管道放在同一个循环内也会出现这样的问题：

那这个bug到底出在哪呢？

很多朋友可能会以为bug处在我们回收进程和关闭管道这里，但却并不是，事实上bug是出在了我们最开始创建管道和进程的地方：

想要弄清楚这个bug，那可得好好的画图分析了：

我们知道父进程再创建一个子进程的时候，子进程会继承父进程的文件描述符表：

如上图，父进程创建出子进程，子进程会继承下来父进程的文件描述符表，只是后来父进程和子进程会关闭对应的读写端。

那么这第一个管道是没什么问题的，但是到第二个管道就会出问题了：

如上图，进程2除了会继承到管道2的读端和写端文件描述符，同时还会多继承一个管道1的文件描述符。

同理类推，如果还有进程3，那进程3除了继承管道3的读写端，还会额外的继承下来管道1和管道2的写端文件描述符。

以此类推，往后每一个新进程都会继承前面创建的所有管道的写端。

那这又会造成什么后果呢？根据我们之前总结的匿名管道的特性：如果写端关闭，读端会一直读，读端0表示读到文件结尾，但是就拿上面的情况来说，如果我们是吧关闭文件描述符和回收子进程的工作放在同一个循环中：

那么：

就相当于只是关闭了管道1的一个写端(在父进程的文件描述符表)，但其实还有一个写端是子进程2还没有关闭，所以子进程1的读端就不会读到0，也就不会退出：

那要怎么解决这个问题呢？

其实我们上面写的先循环关闭管道，再循环回收进程的方法就是一种解决方案：

这是因为，无论我们有多少个管道被创建出来，最后一个管道都是只有一个写端指向它，那么虽然我们现在是从前往后关闭父进程的写端，但最终都是会关闭最后一个管道的写端：

如上图，管道2的唯一写端被关闭了，子进程2就会从管道中读到0，那么子进程就退出了：

而子进程2退出了，子进程2的所有文件描述夫也就关闭了，也就导致了管道1的写端也全关闭了，那么子进程1也就读到0退出。

所以这是一个”从前往后“再从后往前的过程，以此类推，以后我们有十个八个管道也是这样子。

而经过以上分析，我们的第二种方案也就出来了，就是从后往前遍历channels数组，然后在一个循环中先关闭文件描述符，再等待子进程：

运行结果：

但是不知道大家有没有感觉到别扭，就是我们新建的子进程为什么要继承这些额外的，没有必要的文件描述符呢？为什么不能把多余的文件描述符都关闭呢？

那就要引出我们的第三种方案了，这也是我觉得最优的一种方案，那就是将每个子进程多余的文件描述符都管掉！

具体做法：

因为父进程的文件描述符表中是保存着所有的管道的写端文件描述符的，所以我们可以将父进程中所有的管道写端文件文件描述符拷贝到一个数组中，那么在每创建一个一个子进程之后就将该数组中旧的文件描述符关掉即可，最后再由父进程将新管道的写端文件描述符加入到这个数组中：

这样，就算我们吧关闭和回收放在一个循环中也是OK的：

完整代码：

#include <iostream>
#include <vector>
#include <cstring>
#include <string>
#include <unistd.h>
#include <fcntl.h>
#include <cassert>
#include <functional>
#include <ctime>
#include <sys/types.h>
#include <sys/wait.h>
using namespace std;
const int num = 5;
int channel_number = 1; // 管道的编号，之后构成管道的名称
typedef function<void()> task;// 任务
void Play_Music() {cout << "子进程" << getpid() << "正在执行播放音乐的任务……" << endl;
}
void Play_Video() {cout << "子进程" << getpid() << "正在执行播放视频的任务……" << endl;
}void Print_Log() {cout << "子进程" << getpid() << "正在执行打印日志的任务……" << endl;    
}void Download() {cout << "子进程" << getpid() << "正在执行下载任务……" << endl;    
}// channel结构体类型
struct channel {// 构造函数channel(int fd, pid_t id) :ctrfd(fd),workid(id){name = "channel-" + to_string(channel_number++);}int ctrfd; // 控制该管道的文件描述符pid_t workid; // 链接该管道的子进程idstring name; // 管道的名称
};int main() {vector<channel> channels; // 管道数组vector<task> tasks; // 任务数组// 初始化任务数组tasks.push_back(Play_Music);tasks.push_back(Play_Video);tasks.push_back(Print_Log);tasks.push_back(Download);vector<pid_t> oldfd; // 保存旧的文件描述符// 创建进程，创建管道for (int i = 0; i < num; i++) {// 创建管道int pipefd[2] = { 0 };int res = pipe(pipefd);assert(res == 0);// 创建进程pid_t id = fork();assert(id != -1);if (id == 0) {// 先关闭所有旧的文件描述符if (!oldfd.empty()) {for (auto fd : oldfd) {close(fd);}}// 子进程close(pipefd[1]); // 子进程读，子进程关闭写端while (true) {int index = 0;int n = read(pipefd[0], &index, sizeof(index));if (n == sizeof(index)) {// 读取成功，执行任务tasks[index]();} else if (n == 0) {// 说明写端已经关闭，则子进程停止break;}}exit(0); // 让子进程执行完就退出}// 父进程close(pipefd[0]); // 父进程写，父进程关闭读端channels.push_back(channel(pipefd[1], id));oldfd.push_back(pipefd[1]);}// 种一颗随机数种子srand(time(NULL) * getpid());const int T = 10; // 让父进程选择T次for (int i = 0; i < T; i++) {int channeli = i % num; // 管道下标int taski = rand() % tasks.size(); // 任务下标channel& c = channels[channeli];// 向管道发送下标write(c.ctrfd, &taski, sizeof(taski));sleep(1);}// 关闭管道，回收子进程// 方案一// for (auto& c : channels) {//     close(c.ctrfd);//     cout << "close " << c.name << " sucess……" << endl;// }// for (auto& c : channels) {//     pid_t rid = waitpid(c.workid, nullptr, 0);//     if (rid == c.workid) {//         cout << "wait child process:" << rid << " sucess……" << endl;//     }// }// // 方案二// for (int i = channels.size() - 1; i >= 0; i--) {//     close(channels[i].ctrfd);//     cout << "close " << channels[i].name << " sucess……" << endl;//     pid_t rid = waitpid(channels[i].workid, nullptr, 0);//     if (rid == channels[i].workid) {//         cout << "wait child process:" << rid << " sucess……" << endl;//     }// }// 方案三(配合oldfd)for (auto& c : channels) {close(c.ctrfd);cout << "close " << c.name << " sucess……" << endl;pid_t rid = waitpid(c.workid, nullptr, 0);if (rid == c.workid) {cout << "wait child process:" << rid << " sucess……" << endl;}        }return 0;
}