Linux_进程池

1、进程池基本逻辑

2、实现进程池框架

3、文件描述符的继承

4、分配任务给进程池

5、让进程池执行任务

6、回收子进程

7、进程池总结

结语

前言：

在Linux下，进程池表示把多个子进程用数据结构的方式进行统一管理，在任何时候都可以对进程池里的子进程进行任务发放，即进程池可以实现并发执行流，能够同时执行多个任务，相比于单进程单一执行流，进程池在处理多任务的效率上有了显著提升。

1、进程池基本逻辑

进程池之所以能够实现多任务的并发执行，是因为进程池本质是进程间通信（即主进程以通信的形式向进程池里的进程发送任务），并且进程池大部分是由父子进程实现的，所以可以使用匿名管道来实现进程池，说到匿名管道就离不开接口pipe，该接口介绍如下：

#include <unistd.h>//pipe所需要的头文件//传一个数组给到pipe，pipe调用成功时返回0，失败返回-1
//调用成功pipefd数组的第一个元素是读的下标，第二个元素是写的下标
int pipe(int pipefd[2]);

有了此接口就能够搭配fork接口实现父进程与多个子进程的通信了，上面提到进程池是由数据结构对子进程进行管理的，因此我们需要一个数据结构来方便控制子进程，并且可以让父进程通过该数据结构来调度子进程，所以可以定义一个类来描述子进程，代码如下：

class process
{
public:process(int id, const string &name, int fd): id_(id), name_(name), fd_(fd){}~process(){}public:pid_t id_;    // 子进程idstring name_; // 子进程名称int fd_;      // 控制子进程的管道写端(真正控制子进程就是写端的文件描述符)
};

可以理解为创建一个子进程就用process类来描述他，因此创建10个子进程就会有10个process实例化的对象，然后对这10个进行管理即对10个子进程进行管理，可以把这10个对象放到vector内，vector就是管理进程池的数据结构。

至此有了上面的概念，就可以搭建基本的进程池框架了，但是要注意一点，即父进程只读，子进程只写，因此要关闭对应的文件描述符，因为管道是半双工通信，只能一边写一边读，示意图如下：

进程池思路：利用for循环10次，循环调用pipe接口和fork接口，这样就能创建十个子进程，并且每个子进程都有一个匿名管道来与父进程进行通信。当父进程关闭了3号文件描述符后，下一次父进程再次调用pipe接口时，依旧是3号为读端，这样一来创建的子进程统一读端都是3号（因为子进程继承父进程的文件描述符，所以子进程继承了3号读端描述符），子进程的一致性就有了。

2、实现进程池框架

用代码实现进程并简单测试，代码如下：

#include <iostream>
#include <unistd.h>
#include <vector>
#include <error.h>
#include <string>
#include <sys/types.h>using namespace std;
#define N 10class process//描述单个进程的类
{
public:process(int id, const string &name, int fd): id_(id), name_(name), fd_(fd){}~process(){}public:pid_t id_;    // 子进程idstring name_; // 子进程名称int fd_;      // 控制子进程的管道写端(真正控制子进程就是写端的文件描述符)
};int main()
{vector<process> vpr;//管道子进程的数据结构vector<int> oldfd;//记录写端的文件描述符for (int i = 0; i < N; i++)//创建十个子进程的进程池{int pipefd[2];int n = pipe(pipefd);//父进程创建管道if (n == -1){perror("pipe");exit(-1);}// 创建子进程pid_t id = fork();//子进程执行流if (id == 0){for(auto num:oldfd) close(num);//关闭新子进程继承父进程之前的写端close(pipefd[1]);//因为子进程只负责读数据，所以关闭子进程对管道的写端dup2(pipefd[0], 0); // 重定向，这一步只是方便后续的测试exit(0);}// 父进程执行流close(pipefd[0]);//父进程只写，因此关闭父进程对管道的读端string pro_name = "子进程" + to_string(i);vpr.push_back({id, pro_name, pipefd[1]});//把子进程的信息插入到数据结构中oldfd.push_back(pipefd[1]);//记录父进程新打开的写端}for (auto &num : vpr)//验证进程池里的进程{cout << num.name_ << ":" << num.id_ << ":" << num.fd_ << endl;}return 0;
}

运行结果：

从测试结果可以发现，确实生成了十个子进程，并且可以通过vector找到他们，但是上述代码中多创建了一个vector<int> oldfd，这个vector是做什么的呢？

3、文件描述符的继承

我们都知道一个子进程是会继承父进程的PCB结构体的，自然也会继承PCB结构体里的所有内容，而文件描述符就是PCB结构体里的内容之一，所以文件描述符理应被子进程继承，那么在进程池中就会面临这样一个问题：虽然子进程关闭了写端，但是父进程的写端是会越来越多的，而每次创建的子进程只关闭新的写端，会导致新创建的子进程继承了父进程之前打开的写端，并且这些写端没有得到关闭，具体示意图如下：

所以随着越来越多的管道被创建，后续创建的子进程会有大量的写端被打开，并且他们都是指向前面子进程的管道，因此需要用vector记录每一次父进程新打开的写端，因为这些新打开的写端也会拷贝到子进程中，所以在子进程中遍历该vector就能关闭子进程继承而来的写端，这就是vector<int> oldfd的作用。

4、分配任务给进程池

有了上述的进程池框架，接下来就可以对进程池中的每个进程分配任务了，再次之前可以先定义一个任务列表，用函数指针的方式将这些任务用vector管理起来，表示进程池即将处理的任务，任务列表如下：

typedef void (*task)();
vector<task> vt; // 任务队列void task1()
{cout << "检测当前角色健康状态" << endl;
}void task2()
{cout << "检测当前角色物品补给" << endl;
}void task3()
{cout << "检测当前角色生命值" << endl;
}void task4()
{cout << "检测当前角色法力值" << endl;
}void creator_task(vector<task> &vt)//把任务插入到任务队列中
{vt.push_back(task1);vt.push_back(task2);vt.push_back(task3);vt.push_back(task4);
}

有了任务列表后父进程就可以分配任务了，因为任务列表本身是一个vector，并且里面存放的是函数指针，因此父进程给子进程传递vector的下标，这个过程就是任务的派发，子进程拿到下标就可以拿到vector的元素并且调用，这个过程就是任务的执行，父进程派发任务的代码如下：

//父进程开始派送任务cout<<"主进程开始给子进程分配任务"<<endl;sleep(2);for (int i = 0; i < 5; i++){int proc_num = rand()%N;//随机子进程-vpr下标int task_num = rand()%4;//随机任务write(vpr[proc_num].fd_,&task_num,sizeof(int));//分配任务的核心就是进程间通信sleep(1);}

从上述代码中可以发现，让进程池执行任务的本质就是父进程通过调用write函数传递任务列表的下标给到子进程，这就是为什么进程池是通过进程间通信实现的。

5、让进程池执行任务

执行任务主要是子进程的工作，所以在子进程的执行流中还要添加一个等待任务的动作，因为进程池的本质是进程间通信，所以子进程等待任务的动作就是调用read函数，等父进程往匿名管道中写数据（等待任务就是read函数的阻塞），子进程拿到这些数据就可以执行对应的任务了，下面是子进程等待任务的代码：

void chlid_go()//让子进程执行任务
{int task = 0;while (true){int n = read(0, &task, sizeof(int));//读取的内容就是任务列表的下标if (n > 0){cout << "处理该任务的进程是：" << getpid() << ":";vt[task]();//根据读取到的下标去调用任务列表里的函数指针}elsebreak;}
}

把该函数填写到子进程的执行流中，就能让子进程执行任务了，代码如下：

#include <iostream>
#include <unistd.h>
#include <vector>
#include <error.h>
#include <string>
#include <sys/types.h>
#include <time.h>using namespace std;
#define N 10typedef void (*task)();
vector<task> vt; // 任务队列//自定义任务列表
void task1()
{cout << "检测当前角色健康状态" << endl;
}void task2()
{cout << "检测当前角色物品补给" << endl;
}void task3()
{cout << "检测当前角色生命值" << endl;
}void task4()
{cout << "检测当前角色法力值" << endl;
}void creator_task(vector<task> &vt)
{vt.push_back(task1);vt.push_back(task2);vt.push_back(task3);vt.push_back(task4);
}//子进程执行任务
void chlid_go()
{int task = 0;while (true){int n = read(0, &task, sizeof(int));//读取的内容就是任务列表的下标if (n > 0){cout << "处理该任务的进程是：" << getpid() << ":";vt[task]();//根据读取到的下标去调用任务列表里的函数指针}elsebreak;}
}class process//描述单个进程的类
{
public:process(int id, const string &name, int fd): id_(id), name_(name), fd_(fd){}~process(){}public:pid_t id_;    // 子进程idstring name_; // 子进程名称int fd_;      // 控制子进程的管道写端(真正控制子进程就是写端的文件描述符)
};int main()
{creator_task(vt);srand(time(0));vector<process> vpr;//管道子进程的数据结构vector<int> oldfd;//记录写端的文件描述符for (int i = 0; i < N; i++)//创建十个子进程的进程池{int pipefd[2];int n = pipe(pipefd);//父进程创建管道if (n == -1){perror("pipe");exit(-1);}// 创建子进程pid_t id = fork();//子进程执行流if (id == 0){for(auto num:oldfd) close(num);//关闭新子进程继承父进程之前的写端close(pipefd[1]);//因为子进程只负责读数据，所以关闭子进程对管道的写端dup2(pipefd[0], 0); // 重定向，这一步只是方便后续的测试chlid_go();exit(0);}// 父进程执行流close(pipefd[0]);//父进程只写，因此关闭父进程对管道的读端string pro_name = "子进程" + to_string(i);vpr.push_back({id, pro_name, pipefd[1]});//把子进程的信息插入到数据结构中oldfd.push_back(pipefd[1]);//记录父进程新打开的写端}for (auto &num : vpr)//验证进程池里的进程{cout << num.name_ << ":" << num.id_ << ":" << num.fd_ << endl;}//父进程开始派送任务cout<<"主进程开始给子进程分配任务"<<endl;sleep(2);for (int i = 0; i < 5; i++){int proc_num = rand()%N;//随机子进程-vpr下标int task_num = rand()%4;//随机任务write(vpr[proc_num].fd_,&task_num,sizeof(int));//分配任务的核心就是进程间通信sleep(1);}return 0;
}

运行结果：

至此就完成了给进程池里的进程随机派发任务的实现。

6、回收子进程

上述代码结束后没有对子进程做任何的等待工作，但是结果也是正确的，原因就是当父进程退出后，会关闭父进程所有对匿名管道的写端，写端一关闭，则匿名管道的读端就会读到文件末尾，因此read会返回0，在上面代码中当read返回0时就会跳出循环，从而继续往下执行直到exit退出当前子进程，所以父进程不等待子进程则也不会导致孤儿进程问题（因为父进程退出则子进程一定也会退出）。

但是为了保证代码、逻辑的完整性，最好还是写一个专门关闭写端和等待子进程的函数加到上述代码的末尾处，代码如下：

for (int i = 0; i < vpr.size(); i++){close(vpr[i].fd_);//关闭父进程写端waitpid(vpr[i].id_, nullptr, 0);//等待关闭写端的对应子进程cout<<"等待子进程:"<<vpr[i].name_<<endl;sleep(1);}

测试结果：

7、进程池总结

1、进程池通过匿名管道进行父子进程通信而实现的。

2、进程池控制子进程的策略是通过父进程对匿名管道的写端文件描述符，一个写端对应一个子进程。

3、注意文件描述符继承的问题，从逻辑上来说子进程要关闭继承父进程的写端文件描述符，即一个子进程只留下对应匿名管道的读端，而父进程要关闭自己的读端。

4、默认无其他文件描述符，则子进程的读端始终是3号（因为父进程每次都会关闭3号，导致下一次pipe还是3号为读端），并且所有子进程的读端文件描述符是一样的，父进程的写端从4号开始按顺序往下排。