Linux:简易进程池编写

设计概念

进程池，即我们可以预先创建一堆子进程和对应的管道。等父进程有任务时派发给子进程工作。这样就可以节省开辟进程的花销：

当没有任务时，即管道为空。那么子进程就是阻塞状态不会影响其他进程工作效率。

但是注意我们要将任务均衡地派发给子进程，即实现负载均衡

Channel

首先我们要设计一个方案让父进程能统一管理与子进程之间的管道，最好的方法就是封装一个类。
那么类的成员变量必然要有管道的写端fd、子进程的pid还可以有给子进程的编号。

classChannel{public:Channel(intwfd,pid_t id,conststd::string&name):_wfd(wfd),_subprocessid(id),_name(name){}intGetWfd(){return_wfd;}pid_tGetPidProcessId(){return_subprocessid;}std::stringGetName(){return_name;}//关闭写端voidColseChannel(){close(_wfd);}//等待子进程voidWait(){pid_t rid=waitpid(_subprocessid,nullptr,0);if(rid>0){std::cout<<"wait"<<rid<<"success!"<<std::endl;}}~Channel(){}private:int_wfd;pid_t _subprocessid;std::string _name;};

初始化

接下来我们要初始化Channel数组。
根据要创建的worker个数num，我们可以写一个简单的for循环创建管道和子进程。由子进程关闭写端，父进程关闭读端，即可。
但是我们要注意一个细节，这里以创建两个子进程举例，这里先创建第一个worker：

然后分别关闭读写端：

此时我们创建第二个worker：

然后分别关闭读写端：

这时我们发现第二个子进程对第一个管道的写端并没有关闭，此时有可能造成父进程最后无法结束进程。
因此我们在创建新的worker时，要将前面的worker的写端关闭：

voidCreatChannelAndSub(intnum,std::vector<Channel>*channels,task_t task){for(inti=0;i<num;i++){intpipefd[2]={0};intn=pipe(pipefd);//创建管道失败if(n<0)exit(1);//创建子进程pid_t id=fork();if(id==0){//非空就要关闭前面的写端if(!channels->empty()){for(auto&channel:*channels)channel.ColseChannel();}close(pipefd[1]);//重定向到标准输入dup2(pipefd[0],0);//回调函数task();close(pipefd[0]);exit(0);}//父进程std::string channel_name="Channel-"+std::to_string(i);close(pipefd[0]);channels->emplace_back(pipefd[1],id,channel_name);}}

创建任务

我们来实现不同的任务以分配给子进程，首先重命名下函数指针用以实现回调函数：

随意实现三个简单的任务：

voidPrint(){std::cout<<"I am print task"<<std::endl;}voidDownLoad(){std::cout<<"I am download task"<<std::endl;}voidFlush(){std::cout<<"I am flush task"<<std::endl;}

创建回调表：

分配任务：

intSelectTask(){returnrand()%TaskNum;}

执行任务：

voidExcuteTask(intnumber){if(number<0||number>2)return;tasks[number];}

子进程工作

以及有了上面的一系列任务，我们是时候给子进程工作了：

voidwork(){while(true){intcommand=0;intn=read(0,&command,sizeof(command));if(n==sizeof(int)){std::cout<<"pid is:"<<getpid()<<"handler task"<<std::endl;ExcuteTask(command);}//写端关闭if(n==0){std::cout<<"sub process:"<<getpid()<<" quit"<<std::endl;break;}}}

轮询方案

前面提到我们要实现负载均衡，这里可以简单实现为轮询。即轮流给子进程派送任务：

intNextChannel(intchannelnum){staticintnext=0;intchannel=next;next++;next%=channelnum;returnnext;}

分配工作

子进程的工作也有了，轮询方案也有了，就可以给子进程正是分配工作了：

voidSendTaskCommand(Channel&channel,inttaskcommand){write(channel.GetWfd(),&taskcommand,sizeof(int));}voidctrlProcessOnce(std::vector<Channel>&channels){sleep(1);//挑选任务inttaskcommand=SelectTask();//挑选信道和进程intchannel_index=NextChannel(channels.size());//发送任务SendTaskCommand(channels[channel_index],taskcommand);std::cout<<std::endl;std::cout<<"taskcommand:"<<taskcommand<<"channel:"<<channels[channel_index].GetName()\<<"sub process:"<<channels[channel_index].GetPidProcessId()<<std::endl;}voidctrlProcess(std::vector<Channel>&channels,inttimes=-1){if(times>0){while(times--)ctrlProcessOnce(channels);}else{while(true)ctrlProcessOnce(channels);}}

关闭子进程和管道

有了前面关闭子进程所有写端的处理，我们能直接关闭子进程的读端进而使子进程退出，对子进程等待即可：

voidCleanUpChannel(std::vector<Channel>&channels){for(auto&channel:channels){channel.ColseChannel();channel.Wait();}}

Main

接下来就是在main函数里安排代码执行的顺序,我们还可以通过选项的形式控制生成的子进程数量：

intmain(intargc,char*argv[]){if(argc!=2){std::cerr<<"Usage:"<<argv[0]<<" processnum"<<std::endl;}intnum=std::stoi(argv[1]);std::vector<Channel>channels;//1.创建子进程和信道CreatChannelAndSub(num,&channels,work);//2.控制子进程ctrlProcess(channels,5);//3.回收资源CleanUpChannel(channels);return0;}

来尝试运行代码：

完整代码