【Linux 篇】数字世界的通信管道 —— 匿名管道与进程池深度实战解析

于此止？：个人主页

❄️个人专栏传送门：《Linux篇》 《QT百日筑基篇》 《算法篇》

《C++与STL剖析篇》 《数据结构篇》 《Python基础篇》

⭐️纵有狂风拔地起，我亦乘风破万里⭐️

匿名管道与进程池

目录/索引：

目录

匿名管道与进程池

目录/索引：

一、进程间通信的背景

二、进程间通信介绍

三、怎么进行通信？

四、进程间通信的发展

五、进程间通信的分类

1.管道：

2.system V IPC

3.POSIX IPC

六、管道

什么是管道？

命令行中管道的使用

七、匿名管道

1.通信方式

2.通信的原理

3.创建匿名管道

4.利用匿名管道进行进程间的通信

任务1.管道的创建。

任务2.创建子进程。

任务3：对文件的读写操作。

源码：

八、匿名管道的5中特性

九、管道的四种通信情况

十、进程池的编写

验证我们管道的容量：

文件类型：

进程池的结构

进程池代码剖析：

结构框架

成员变量：

进程池的数量

创建进程池：

子进程的工作

运行进程池：

信道的选择：

进程池的销毁：

Bug？

进程池的源码：

ProccessPool.hpp

tasktable.hpp:

test.cc

一、进程间通信的背景

Linux 中每个进程拥有独立虚拟地址空间，进程内存相互隔离，无法直接访问对方数据。

实际开发经常需要多进程协同工作、分发任务、交换数据，因此诞生了进程间通信（IPC）。

匿名管道是经典轻量 IPC 方式，适用于存在亲缘关系的父子进程，非常适合用来实现进程池任务分发。

Linux 进程具备内存隔离特性，进程之间不能直接共享数据。想要实现多进程协同、构建进程池调度任务，就需要进程间通信。匿名管道作为简洁高效的通信方案，常被用于父子进程之间的数据传输。

什么是通信呢？

通信就是将信息从发送方传递到接收方的过程。 分为模拟通信、数字通信；也可分为有线通信、无线通信。 一套通信系统包含信源、发送设备、信道、接收设备、信宿，传输过程会受到噪声干扰。日常手机、网络、蓝牙、光纤等都属于通信应用

二、进程间通信介绍

1.数据传输：一个进程需要将消息发送给另一个进程。

2.资源共享：多个进程共享同一份资源。

3.通知事件：一个进程需要向另一个或者另一组进程发消息，通知他们发生了某些事并执行某种任务（子进程退出时会将自己的退出信息告诉父进程）。

4.进程控制：有些进程希望自己去控制另一个进程（Debug进程）此时控制进程能够拦截另一个进程的所有陷入和异常 ，并能够通知他们他的状态发生改变。

三、怎么进行通信？

我们先来说一说进程间通信的本质：让不同的进程看到同一份的资源。（内存）这分资源由我们os进行开辟空间。

os会给我们进行提供对应的系统调用的接口。我们去进行调用这块接口去完成对应的操作。

os在设计这块的接口的时候会统一设置这块的接口来进行控制这份资源。

四、进程间通信的发展

1.管道。

2.system V进程间的通信。

3.POSIX进程间的通信。

五、进程间通信的分类

1.管道：

匿名管道pipe

命名管道

2.system V IPC

-system V消息队列

-system V共享内存

-system V信号量

3.POSIX IPC

-消息队列

-共享内存

-信号量

-互斥锁

-条件变量

-读写锁

六、管道

什么是管道？

管道是Unix最古老的进程间通信的形式。

我们把从一个进程连接到另一个进程的一个数据流叫做管道。

管道是内核级的。

命令行中管道的使用

我们下面的"|"就代表管道，这个命令的意思是我们首先使用ls -l打印出当前目录的所有的文件，在使用我们的grep过滤出我们想要的文件。这样我们对应的命令行就能给我们打印出来对应的文件信息，下面就是我们管道的一个简单的使用，以后我们会用于ps等命令中，让我们更快更好的过滤出我们想要的信息。

ls -l | grep code.c

七、匿名管道

匿名管道是我们当时基于现有的技术（文件操作）直接进行通信。

1.通信方式

1.基于文件的管道通信。

2.基于system的本机通信。

2.通信的原理

匿名管道通常是我们父子间的通信。

我们对应创建子进程时我们的子进程会拷贝父进程的数据，进行操作。拷贝到的内容是和父进程是一样的，只有我们进行修改的时候才会发生对应的写实考呗。那么这样我们在创建子进程的时候成功让我们的不同的进程看到同一份资源。那么我们就可以进行对应的进程间的通信了。

下图有个小问题我们在创建对应的子进程的时候，我们的文件还要在重新的进行深拷贝吗？

答案是不用的，我们对应的struct_file这个表储存的是文件的描述符，我们只要通过对应的文件描述符就能找到对应的文件， 我们后面他会不会打开文件和我们的进程没有任何关系。因为我们前面属于进程管理，我们后面的一部分属于文件管理。所以我们创建子进程时会将文件描述符表给我们进行拷贝一份，这次的拷贝属于是浅拷贝。

我们可以先创建一个管道， 然后我们的创建一个子进程，这是我们的父子进程的读写端都链接到这个管道这里， 这时候我们将父进程的读端关闭， 我们将子进程的写端进行关闭，这样我们就能让父进程进行向管道中写内容，我们的子进程向管道中读内容这样我们就能完成对应的进程间的通信了，并且看到了同一份资源（管道）（这份资源是我的子进程继承了父进程的资源）。同理我们的父进程管写，子进程关读也是可以的。

3.创建匿名管道

我们这边使用pipe进行匿名管道的创建我们这边使用pipe进行匿名管道的创建

我们首先看一下我们的手册：

man 2 pipe

我们这需要传过去一个int类型的数组， 他会将我们对应创建好的管道的2文键描述符该我们进行返回，2个一个读端，一个写端。我们对应创建出来的管道没有名字因此叫做匿名管道。

返回值：

成功返回0，失败返回-1，错误码被设置。

4.利用匿名管道进行进程间的通信

任务1.管道的创建。

ok，这里我们在看一下文档。头文件<unistd.h>,要先创建一个fd有两个空间的数组。

然后我们调用pipe。

我们可以将我们对应管道的文件描述符给打印出来对应是几号文件描述符。

int main() { int status[2] = {0};//创建对应的数组 int ret_pipe = pipe(status);//创建管道 if (ret_pipe < 0) {//判断是否成功 perror("pipe false"); exit(1); } std::cout<< status[0]<<":"<<status[1]<<std::endl;//判断对应的数组的内容 return 0; }

注意这里申请的是3号和4号， 这两个号其中3号是读端，我们可以这样进行记忆， status对应的status[0]这个0像一个嘴巴在读，所以他就是我们对应的读端。也就是我们对应的3号文件描述符。

我们对应的status[1]像一支笔，我们要写一些东西，也就代表我们的4号文件描述符。他的文件描述符也是支持就近原则的，谁小有限先择谁。

任务2.创建子进程。

下一步我们就要创建对应的子进程来完成对应的任务了：

我们这里是让我们的子进程进行写入，让我们的父进程进行读取。

所以我们子进程要进行写入，所以我的子进程要关闭对应的读端。我们将我们子进程要干的活设置为对应的child函数进行封装。

我们的父进程要进行对应的读取，我们的父进程关闭对应的写端，我们让父进程的工作用函数parent来进行封装。所以我们子进程要进行对应的写入，父进程要进行我们对应的读取这不就是我们的文件操作了吗。

int main() { int status[2] = {0}; int ret_pipe = pipe(status); if (ret_pipe < 0) { perror("pipe false"); exit(1); } // std::cout<< status[0]<<":"<<status[1]<<std::endl; // 创建对应的进程： pid_t id = fork(); if (id == 0) { close(status[0]); child(status[1]); close(status[1]); } close(status[1]); parent(status[0]); waitpid(id, 0, 0); close(status[0]); return 0; }

任务3：对文件的读写操作。

我们子进程向文件写就行了，我们的父进程一直的读取就行了。

所以现在我们就是基于文件进行操作：

void child(int wfd) { char buffer[1024]; int cnt = 0; while (1) { snprintf(buffer, sizeof(buffer), "I am child pid:%d,cnt: %d", getpid(), cnt++); write(wfd, buffer, strlen(buffer)); sleep(1); } } void parent(int wfd) { char buffer[1024]; while(1) { buffer[0]=0; ssize_t n=read(wfd, buffer,sizeof(buffer)-1); if(n>0) { buffer[n]=0; std::cout<<"child say:"<< buffer<<std::endl; } sleep(1); } }

ok我们的一个匿名管道就做好了我们进行运行一下，通信：

我们也发现他是1s中进行打印一条消息。

源码：

#include <cstdio> #include <cstring> #include <iostream> #include <sys/types.h> #include <sys/wait.h> #include <unistd.h> void child(int wfd) { char buffer[1024]; int cnt = 0; while (1) { snprintf(buffer, sizeof(buffer), "I am child pid:%d,cnt: %d", getpid(), cnt++); write(wfd, buffer, strlen(buffer)); sleep(1); } } void parent(int wfd) { char buffer[1024]; while(1) { buffer[0]=0; ssize_t n=read(wfd, buffer,sizeof(buffer)-1); if(n>0) { buffer[n]=0; std::cout<<"child say:"<< buffer<<std::endl; } sleep(1); } } int main() { int status[2] = {0}; int ret_pipe = pipe(status); if (ret_pipe < 0) { perror("pipe false"); exit(1); } // std::cout<< status[0]<<":"<<status[1]<<std::endl; // 创建对应的进程： pid_t id = fork(); if (id == 0) { close(status[0]); child(status[1]); close(status[1]); exit(0); } close(status[1]); parent(status[0]); waitpid(id, 0, 0); close(status[0]); return 0; }

八、匿名管道的5中特性

我们通过上面的试验也能看出来管道的这5中特性。

1.匿名管道只能进行有血缘关系的进程才能通信。（兄弟之间也是可以的）

2.管道文件带有同步机制。（你写一条我读一条）

3.管道是面向字节流的。

4.匿名管道是单项通信的。（一端写一端读）

-一个写一个读 ——半双工

-两个都能写都能读 —— 全双工

5.管道的生命周期是随进程的。（底层有对应的引用计数）

九、管道的四种通信情况

1.写慢，读快 —— 读端阻塞，等写端写，写一条读一条。

我们只要把对应的写段sleep上几秒就可以。其余代码保持不变。（基于上面的源码）

void child(int wfd) { char buffer[1024]; int cnt = 0; while (1) { snprintf(buffer, sizeof(buffer), "I am child pid:%d,cnt: %d", getpid(), cnt++); write(wfd, buffer, strlen(buffer)); sleep(3); } }

2.写快，读慢 —— 写端一直写，读端一读读出一大串的内容。写完了，写端阻塞，等读。

我们只要把对应的读端sleep上几秒就可以。其余代码保持不变。（基于上面的源码）

void parent(int wfd) { char buffer[1024]; while(1) { buffer[0]=0; ssize_t n=read(wfd, buffer,sizeof(buffer)-1); if(n>0) { buffer[n]=0; std::cout<<"child say:"<< buffer<<std::endl; sleep(2); } sleep(1); } }

3.写关，读不关 —— 写端会默认读到最后一个字符返回0，读端也要进行关闭。

我们让程序写一次就进行关闭（break，外面会调用close进行关闭），我们的写端关闭我们的读端默认就是读到了文件的结尾，read返回值为0，其余代码保持不变。（基于上面的源码）

void child(int wfd) { char buffer[1024]; int cnt = 0; while (1) { snprintf(buffer, sizeof(buffer), "I am child pid:%d,cnt: %d", getpid(), cnt++); write(wfd, buffer, strlen(buffer)); break; } } void parent(int wfd) { char buffer[1024]; while(1) { buffer[0]=0; ssize_t n=read(wfd, buffer,sizeof(buffer)-1); if(n>0) { buffer[n]=0; std::cout<<"child say:"<< buffer<<std::endl; } else if(n ==0) { std::cout<<"w退出了，我也退出"<<std::endl; break; } } }

4.写不关，读关 —— 程序异常， 父进程一直等待回收子进程，子进程一直在哪里写，写满等读。

读一次我们就关闭，我们看到我们读了一次我们的程序就会阻塞在哪里， 我们的子进程没有退出父进程就会阻塞在哪里想要回收子进程。其余代码保持不变。（基于上面的源码）

十、进程池的编写

我们现在编写的进程池是基于我们的匿名管道来实现我们对应的多进程之间的操作的。

验证我们管道的容量：

我们将读端进行关闭运行下面的一个程序：

我们每一次写4个字节，看看一共能写多少。

void child(int wfd) { char buffer[1024]; int cnt = 0; while (1) { write(wfd, &cnt, 4); cnt++; std::cout<<cnt<<std::endl; } }

我们看到了一共16385次那我们计算一下16384 x 4 = 65540 Byte， 65540 / 1024 约等于64个字节。所以我们能进行使用的是64个字节左右。

文件类型：

我们先认识我们对应的文件类型

我们会用到对应的hpp文件， 他是我们.h文件和.cpp文件的合并形式。

我们之前定义.h文件会进行分文件去实现对应的.cpp文件， 而我们.hpp文件定义完可以直接的进行使用。

我们的源文件.cc .cpp .cxx都是我们对应的c++文件的后缀， 我们哪个文件都可以进行使用。

进程池的结构

我们可以创建一个父进程，来进行写， 然后pipe创建很多个管道来进行对应的通信这里设置为5个然后创建5个子进程（fork）来进行读取， 每当我父进程写入的时候我们对应的这5个子进程来进行对应的读取， 我们能设置对应的读写规则。父进程写入通过管道写入到不同的子进程中。

进程池代码剖析：

1.我们要有的结构（类）

首先我们每一个信道要进行创建Channel 就代表一个信道。

其次我们的信道要进行对应的组织起来。方便我们后续的管理。（Grp_Channel ）

然后就是编写我们的进程池的一个专属的类负责提供对应的接口。（Propool ）

结构框架

我们的结构如下：

#include <iostream> class Channel { Channel() { } ~Channel() { } }; class Grp_Channel { Grp_Channel() { } ~Grp_Channel() { } }; class Propool { Propool() { } ~Propool() { } };

成员变量：

那我们对应的成员都有啥呢？

一定有我们的文件描述符fd，我们创建管道一定需要读写所以一定要添加fd。

我们的Grp_Channel就是将我们的Channel进行组织起来。

所以我们可以选择一个数据结构这里选择vector（list， map）都可以。

我们的进成池一定要有一个Grp_Channel类型的变量。

当然还有对应的成员我们后面进行加入。

所以：

class Channel { public: Channel() { } ~Channel() { } private: int fd; int id; std::string name; }; class Grp_Channel { public: Grp_Channel() { } ~Grp_Channel() { } private: std::vector<Channel> _cn; }; class Propool { public: Propool() { } ~Propool() { } private: Grp_Channel _cm; };

进程池的数量

我们这里要设置管道的数量。我们默认是5。也就代表了我们子进程的数量。

#define PROC_MAX 5

创建进程池：

在使用进程池通信之前我们要先创建对应的进程池。

我们的子进程要执行读的工作所以我们的Work，所以我们专们写一个函数来进行执行。

我们的父进程要做的工作就是把我们的文件描述符给保存起来，有助于我们进行管理。

bool Create() { for (int i = 0; i < _num; i++) { int pipefd[2] = {0}; int ret_pipe = pipe(pipefd); if (ret_pipe < 0) { perror("pipe fail"); return false; } pid_t id = fork(); if (id == 0) { close(pipefd[1]); Work(pipefd[0]); close(pipefd[1]); exit(0); } else if (id > 0) { close(pipefd[0]); _cm.Push(id, pipefd[1]); } else { perror("fork fail!"); return false; } } return true; }

void Push(int id, int fd) { _vc.emplace_back(id, fd); }

Channel(pid_t id, int fd) : _id(id), _fd(fd) { name = "我是一个子进程-" + std::to_string(fd) +'-'+std::to_string(id); }

子进程的工作

这里我们的子进程要干什么呢？

当然是读取我们对应的数据数了。这里我们设置一个整数code，表示一个任务码，我们会发出很多的任务码便于我们执行不同的任务。所以我们设置一个code。这里注意如果我们写端不写他就会阻塞在那里。所以刚开始的时候就会进行阻塞。我们的父进程负责发对应的任务码。

void Work(int fd) { while (1) { int code=0; ssize_t ret_r = read(fd, &code, sizeof(code)); if (ret_r < 0) { perror("read fail!"); } else if (ret_r == 0) { std::cout << "w端退出" << std::endl; break; } else { if(ret_r==sizeof(code)) { std::cout<<"子进程[" << getpid<<"]收到了一个退出码"<<std::endl; } else continue; } } }

运行进程池：

这里要注意我们发出了对应的任务码，必须先创建对应的任务

所以这里我们要先写我们有什么的任务。

这里随便给出了三个任务。我们的Register是我们想要向文件中添加什么任务。

Code是我们返回一个随机任务让我们进行执行。Execute这个主要是给我们的子进程进行设置的让我们的子进程的得到对应的任务码的时候去执行对应的函数。

#pragma once #include <iostream> #include <vector> #include <ctime> typedef void (*task_t)(); ////////////////debug///////////////////// void PrintLog() { std::cout << "我是一个打印日志的任务" << std::endl; } void Download() { std::cout << "我是一个下载的任务" << std::endl; } void Upload() { std::cout << "我是一个上传的任务" << std::endl; } ////////////////////////////////////// class TaskManager { public: TaskManager() { srand(time(nullptr)); } void Register(task_t t) { _tasks.push_back(t); } int Code() { return rand() % _tasks.size(); } void Execute(int code) { if(code >= 0 && code < _tasks.size()) { _tasks[code](); } } ~TaskManager() {} private: std::vector<task_t> _tasks; };

所以我们在运行进程池的时候要注意这些任务，我们要先进性注册对应任务。

ProcessPool(int num):_num(num) { _tb.Register(PrintLog); _tb.Register(Download); _tb.Register(Upload); } TaskManager _tb;

这里我们的父进程要先去选择一个任务让我们去执行。

然后选择一个信道让我们去执行。

我们向这个信道去发送这个任务码。

void Proc_Start() { //选择一个任务 int taskcode = _tb.Code(); //选择一个信道来完成任务。 auto&c=_cm.Select(); std::cout<<"选择了一个子进程"<<c.GetName()<<std::endl; //发送对应的任务来完成 c.Send(taskcode); std::cout<<"发送了一个对应的任务码："<<taskcode<<std::endl; }

信道的选择：

我们采用什么方式进行选择，我们的选则是避免忙的忙死，闲的闲死。

1.轮询的方式进行

2.随机选择

3.添加负载指标

所以这里我们原则轮询的方式

Channel& Select()//轮询的方式进行查找 { Channel& ret = _vc[_next]; _next++; _next %= _vc.size(); return ret; }

我们要加入一个_next的成员变量让我们找下一个进行工作的信道。

_next %= _vc.size();为了避免我们进行越界。

我们这里在提供一种随机：

srand(0); int a = rand() % _vc.size(); return _vc[a];

我们选择了这个信道让这个信道去write到管道中，让对应的进程来读取对应的任务码来进行去解决对应的任务。

void Send(int code) { ssize_t n = write(_fd, &code, sizeof(code)); (void)n; }

然后我们基于现在的子进程还要再加入对应的执行函数，让我们执行对应任务码对应的任务。

void Work(int fd) { while (1) { int code=0; ssize_t ret_r = read(fd, &code, sizeof(code)); if (ret_r < 0) { perror("read fail!"); } else if (ret_r == 0) { std::cout << "w端退出" << std::endl; break; } else { if(ret_r==sizeof(code)) { std::cout<<"子进程[" << getpid<<"]收到了一个退出码"<<std::endl; //执行对应的这个任务 _tb.Execute(code); } else continue; } } }

上面我已经写好了一个对应的信道的运行和创建

进程池的销毁：

我们的进程池的销毁主要是由2的东西，1.我们的父进程回收所有的子进程， 2.我们的文件描述符要进行关闭。

这里我们的管道要不要进行销毁呢？

答案不要，因为我们对应的管道是对应的引用计数，当我们的引用技术为0的时候我们的OS回自行进行回收。

我们的进程池只需要调用一次：

void StoProcPool()//终止进程池 { _cm.delete_Stop(); }

Grp_Channel

void start_close() { for(auto& vc : _vc) { vc.Close(); } } void start_wait() { for(auto& vc: _vc) { vc.Wait(); } }

Channel：

我们的id和fd在我们的一个个信道中就有了。所以我们循环进行调用就可以了。

void Close() { close(_fd); } void Wait() { pid_t wi = waitpid(_id,nullptr,0); (void) wi; }

Bug？

当然这个代码有一个很深的BUG？我们可以知道当我们创建一个子进程的时候我们回拷贝他的文件描述符表所以我们下一个创建的子进程文件描述符表中会有指向我们上一个创建管道。以此类推我们的下下一个会指向我们的前两个的管道。

我们可以从下开始进行对应的释放我们对应的表：

//解决方案1：从下向上进行回收 // std::vector<Channel>::reverse_iterator it = _vc.rbegin(); // while(it != _vc.rend()) // { // it->Close(); // std::cout<<"关闭"<<it->GetName()<<std::endl; // it->Wait(); // std::cout<<"回收"<<it->GetName()<<std::endl; // ++it; // }

方案二：

提前进行一次将文件描述符我们对应不需要的进行关闭。在创建的时候我们就把不要的文件描述符进行关闭就可了，我们对应的通道就一定只有我们的2个进程进行指向了。

_cm.start_close();//进程是独立的我们在父进程获取的一些元素文件描述符我们进行关闭

//我们的文件描述符会不会关闭呢？答案是不会的，因为就算我们的父进程先发生push也会发生写实考呗。

for(auto& vc : _vc) { vc.Close(); std::cout<<"关闭"<<vc.GetName()<<std::endl; vc.Wait(); std::cout<<"回收"<<vc.GetName()<<std::endl; }

进程池的源码：

ProccessPool.hpp

#pragma once #include <cstring> #include <iostream> #include <sys/types.h> #include <unistd.h> #include <vector> #include <sys/wait.h> #include "tasktable.hpp" #define CHANNEL_NUM 5 #define READ_MAX 1024 // 先描述 class Channel { public: Channel(pid_t id, int fd) : _id(id), _fd(fd) { name = "我是一个子进程-" + std::to_string(fd) +'-'+std::to_string(id); } ~Channel() { } std::string& GetName() { return name; } pid_t Getpid() { return _id; } int Getfd() { return _fd; } void Send(int code) { ssize_t n = write(_fd, &code, sizeof(code)); (void)n; } void Close() { close(_fd); } void Wait() { pid_t wi = waitpid(_id,nullptr,0); (void) wi; } private: pid_t _id; int _fd; std::string name; }; // 在组织 class AdmChannel { public: AdmChannel() :_next(0){ } void Push(int id, int fd) { _vc.emplace_back(id, fd); } Channel& Select()//轮询的方式进行查找 { Channel& ret = _vc[_next]; _next++; _next %= _vc.size(); return ret; // srand(0); // int a = rand() % _vc.size(); // return _vc[a]; } void PrintCe()//测试 { for(auto& vc:_vc) { std::cout<<vc.GetName()<<std::endl; } } void delete_Stop() { //解决方案二： 从源头上解决，我们的原因是不同的文件描述符指向了同一块区域。我们提供一个close提前进行关闭。 for(auto& vc : _vc) { vc.Close(); std::cout<<"关闭"<<vc.GetName()<<std::endl; vc.Wait(); std::cout<<"回收"<<vc.GetName()<<std::endl; } //解决方案1：从下向上进行回收 // std::vector<Channel>::reverse_iterator it = _vc.rbegin(); // while(it != _vc.rend()) // { // it->Close(); // std::cout<<"关闭"<<it->GetName()<<std::endl; // it->Wait(); // std::cout<<"回收"<<it->GetName()<<std::endl; // ++it; // } } void start_close() { for(auto& vc : _vc) { vc.Close(); } } void start_wait() { for(auto& vc: _vc) { vc.Wait(); } } ~AdmChannel() {} private: std::vector<Channel> _vc; int _next; }; class ProcessPool { public: ProcessPool(int num):_num(num) { _tb.Register(PrintLog); _tb.Register(Download); _tb.Register(Upload); } void Work(int fd) { while (1) { int code=0; ssize_t ret_r = read(fd, &code, sizeof(code)); if (ret_r < 0) { perror("read fail!"); } else if (ret_r == 0) { std::cout << "w端退出" << std::endl; break; } else { if(ret_r==sizeof(code)) { std::cout<<"子进程[" << getpid<<"]收到了一个退出码"<<std::endl; //执行对应的这个任务 _tb.Execute(code); } else continue; } } } bool Create() { for (int i = 0; i < _num; i++) { int pipefd[2] = {0}; int ret_pipe = pipe(pipefd); if (ret_pipe < 0) { perror("pipe fail"); return false; } pid_t id = fork(); if (id == 0) { close(pipefd[1]); _cm.start_close();//进程是独立的我们在父进程获取的一些元素文件描述符我们进行关闭 //我们的文件描述符会不会关闭呢？答案是不会的，因为就算我们的父进程先发生push也会发生写实考呗。 Work(pipefd[0]); close(pipefd[1]); exit(0); } else if (id > 0) { close(pipefd[0]); _cm.Push(id, pipefd[1]); } else { perror("fork fail!"); return false; } } return true; } void Proc_Start() { //选择一个任务 int taskcode = _tb.Code(); //选择一个信道来完成任务。 auto&c=_cm.Select(); std::cout<<"选择了一个子进程"<<c.GetName()<<std::endl; //发送对应的任务来完成 c.Send(taskcode); std::cout<<"发送了一个对应的任务码："<<taskcode<<std::endl; } void Debug() { _cm.PrintCe(); } void StoProcPool()//终止进程池 { _cm.delete_Stop(); } ~ProcessPool() {} private: AdmChannel _cm; int _num; TaskManager _tb; };

tasktable.hpp:

#pragma once #include <iostream> #include <vector> #include <ctime> typedef void (*task_t)(); ////////////////debug///////////////////// void PrintLog() { std::cout << "我是一个打印日志的任务" << std::endl; } void Download() { std::cout << "我是一个下载的任务" << std::endl; } void Upload() { std::cout << "我是一个上传的任务" << std::endl; } ////////////////////////////////////// class TaskManager { public: TaskManager() { srand(time(nullptr)); } void Register(task_t t) { _tasks.push_back(t); } int Code() { return rand() % _tasks.size(); } void Execute(int code) { if(code >= 0 && code < _tasks.size()) { _tasks[code](); } } ~TaskManager() {} private: std::vector<task_t> _tasks; };

test.cc

#include "ProcessPool.hpp" int main() { std::cout<<"你好，进程池"<<std::endl; //创建进程池 ProcessPool pp(CHANNEL_NUM); pp.Create(); //启动进程 int cnt =10; while(cnt--) { pp.Proc_Start(); sleep(1); } pp.StoProcPool(); return 0; }