【进程通信】利用管道创建进程池（结合代码）

文章目录

什么叫进程池
- 进程池的优点
创建进程池
- 代码实现：

什么叫进程池

我们知道，一个进程创建子进程通常是为了让这个子进程去为它完成某个任务。例如我们使用的指令，其实就是bash进程创建子进程让子进程去执行的。但是我们需要考虑这样一个问题：是不是遇到问题之后才创建子进程呢？

频繁的创建和销毁进程都是一项较大的开销，涉及到内存分配、上下文切换等操作。于是我们可以提前创建出一批进程，当有任务要做的时候就从这一批子进程中拿出一个空闲的去执行，一旦某个子进程把任务执行完之后也不立即销毁进程，而是等待下一个任务的来临。

我们把这些预先创建的一批子进程就叫做进程池。我们把进程池中的进程也称为工作进程。

进程池的优点

通过复用已经创建的进程，减少创建和销毁进程的开销，提高了系统资源的利用与率和系统的性能。
动态调整进程池中工作进程的数量。可以让系统根据实际的负载和任务需求，动态的增加或者减少工作进程的数量，以适应不同的工作负载。提高了系统的工作的灵活性。
简化编程。进程池提供了一种高级抽象，隐藏了底层进程管理的细节，使得并发进程更加简单和可靠。

创建进程池

现在我们尝试自己去实现一个进程池。
创建进程好很简单，如何管理这些进程呢？每创建一个进程我们都用一个结构体（或者类）维护，于是对工作进程的管理就变成了对结构体的管理，整个进程池从代码角度上来说就是一个结构体数组。

那我们如何将任务分配到工作进程呢？这个问题的本质其实是在问进程之间通信的方式。于是我们很容易的想到使用管道。父进程通过管道，将任务信息传递给子进程。

下面是使用进程池的一个简单模型：

在这里插入图片描述

代码实现：

Process.cpp文件


#include <iostream>
#include <cerrno>
#include <unistd.h>
#include <string>
#include <vector>
#include <sys/types.h>
#include <wait.h>
#include <sys/stat.h>
#include "Task.hpp"
using namespace std;// master
class Channel // 维护管道信息
{
public:Channel(int wfd, int subprocessid, string name): _wfd(wfd), _subprocessid(subprocessid), _name(name){}int GetWfd() { return _wfd; }int GetProcessId() { return _subprocessid; }string GetName() { return _name; }void CloseChannel(){ // 关闭信道的写端close(_wfd);}void Wait(){ // 关闭工作进程int status = 0;pid_t rid = waitpid(_subprocessid, NULL, 0);if (rid > 0){cout << "wait " << rid << " success" << endl;}}~Channel(){}private:int _wfd;          // 信道以写方式打开的文件描述符int _subprocessid; // 读端进程的pidstring _name;      // 信道的命名
};// 创建信道和子进程
void CreatChannelAndSub(int num, vector<Channel> &Channels, task_t task)
{ // task是一个回调函数for (int i = 0; i < num; i++){// 1.创建管道int pipefd[2];int n = pipe(pipefd);if (n < 0)exit(0);// 创建子进程pid_t pid = fork();if (pid == 0){// 关闭写端close(pipefd[1]);dup2(pipefd[0], 0); // 将管道的读端重定向到标准输入task();             // 子进程执行的任务close(pipefd[0]);exit(0);}// 父进程关闭读端，只用来输出任务close(pipefd[0]);string name = "Channel-" + to_string(i);// 将创建的信道存入输出型参数中Channels.push_back({pipefd[1], pid, name});}
}int NextChannel(int channelnum)
{static int next = 0;int channel = next;next = (next + 1) % channelnum;return channel;
}void SendTask(int taskcommand, Channel &channel)
{write(channel.GetWfd(), &taskcommand, sizeof(taskcommand));
}// 具体任务
void CtrlProcessOnce(vector<Channel> &Channels)
{// 1.选择一个任务int taskcommand = SelectTask();// 2.选择一个空信道int taskchannel = NextChannel(Channels.size());// 3.发送任务编号给信道SendTask(taskcommand, Channels[taskchannel]);cout << "taskcommand: " << taskcommand << " channel: " << Channels[taskchannel].GetName() << " sub process: " << Channels[taskchannel].GetProcessId() << endl;
}// 控制子进程执行任务
void CtrlProcess(vector<Channel> &Channels, int time = -1)
{ // time表示分配多少次任务给子进程去执行，默认是无限if (time == -1){ // 默认while (true){CtrlProcessOnce(Channels);sleep(1);}}else if (time > 0){while (time--){CtrlProcessOnce(Channels);sleep(1);}}
}void CleanUpChannel(vector<Channel> &channels)
{for (auto &it : channels){it.CloseChannel();}for (auto &it : channels){it.Wait();}
}int main(int argc, char *argv[])
{ // 参数列表，argv[1]表示创建的工作进程的数量，即内存池大小if (argc != 2){cerr << "Usage: " << argv[0] << " Processnum" << endl;return 1;}int num = stoi(argv[1]);// 加载任务LoadTask();vector<Channel> Channels; // 进程池// 创建信道和子进程CreatChannelAndSub(num, Channels, work1);// 通过Channel控制子进程CtrlProcess(Channels, 10);// 关闭子进程和信道的写端CleanUpChannel(Channels);return 0;
}

Task.hpp文件


#include <iostream>
#include <ctime>
#include <sys/stat.h>
#include <cstdlib>
#include <unistd.h>
#include <sys/types.h>
#define TASKNUM 3
using namespace std;typedef void (*task_t)(); // task_t函数指针类型void Print()
{ // 任务1cout << "Printf task" << endl;
}void DownLoad()
{ // 任务2cout << "DownLoad task" << endl;
}void Flush()
{ // 任务3cout << "Flush task" << endl;
}task_t tasks[TASKNUM]; // 用来存放函数指针void LoadTask()
{srand(time(nullptr));tasks[0] = Print;tasks[1] = DownLoad;tasks[2] = Flush;
}void ExcuteTask(int n)
{if (n < 0 || n > 2)return;tasks[n]();
}int SelectTask()
{int n = rand() % TASKNUM;return n;
}void work1()
{while (true){int command = 0;int n = read(0, &command, sizeof command); // 读取管道中的任务编号(int)if (n == sizeof(int)){ExcuteTask(command);}else if (n == 0){cout << "sub process: " << getpid() << " quit " << endl;break;}}
}