进程池设计
代码目的头文件代码语言：c++复制
#include#include#include#include#include#include#include#include对子进程操作建立子进程对象并把子进程对象放进数组里代码语言：c++复制
//创建子进程对象class SubEp//endpoint---子进程对象{public:SubEp(pid_t subid,int writefd)//第一个参数是子进程的pid，第二个参数是该子进程读端对于父进程的写端fd:_subid(subid),_writefd(writefd){ char namebuffer[1024]; //第一个参数是表示第几号子进程，第二个参数是子进程的pid，第三个参数是该子进程读端对于的父进程的写端 snprintf(namebuffer,sizeof namebuffer,"process: %d [pid(%d) - fd(%d)]",num++,_subid,_writefd); _name=namebuffer;}public:static int num;string _name;pid_t _subid;int _writefd;//该子进程与父进程匿名管道对于的父进程的写端fd};int SubEp::num=0;子进程对象需要传递两个参数来初始化成员变量_subid和 _writefd。一是子进程的pid二是该子进程读端对应父进程写端的文件描述符fd成员变量num表示是第几个创建出来的子进程，第一个创建出来的子进程为0，使用后++后续子进程的num依次是1，2等等。因此num不能由于出了SubEp对象作用域后被销毁，所以定义为static，变量num生命周期取决于SubEp类的生命周期成员变量 _name用namebuffer初始化，用来标识该子进程的其他成员变量建立子进程需要执行的任务表代码语言：c++复制
//创建父进程给子进程派发的任务列表typedef void(*func_t)();//函数指针类型，函数返回值为voidvoid downloadTask()//模拟下载任务{ cout<<getpid()<<": 下载任务n"<<endl; sleep(1);}void fflushTask()//模拟刷新任务{ cout<<getpid()<<": 刷新任务n"<<endl; sleep(1);}void subscribeTask()//模拟订阅任务{ cout<<getpid()<<": 订阅任务n"<<endl; sleep(1);}//把上面的三种任务load到列表中即让存放函数指针的vector的各个指针能够指向上面的函数，为了后面方便调用void loadTaskFunc(vector*out){ assert(out);//vector创建成功 out->push_back(downloadTask); out->push_back(fflushTask); out->push_back(subscribeTask);}子进程需要执行的任务都是函数对象，建立一个对象是函数指针的数组out，通过loadTaskFunc函数把任务函数尾插到数组out里，然后通过输出型参数返回。创建子进程和父进程通信的管道，并且让子进程阻塞读取代码语言：c++复制
void CreateSubProcesses( vector*subs,vector& funcMap){ vector deleteFd;//创建子进程并且创建好父进程与各个子进程通信的管道int fds[2];for(size_t i=0;i<PROCESS_NUM;i++)//创建子进程{ int n=pipe(fds);//建立父子间进程的匿名管道--建立成功返回0，建立失败返回-1 assert(n==0);//判断管道是否建立成功 (void)n; pid_t id=fork();//创建子进程 if(id==0) { for(size_t i=0;i=0 && taskcodepush_back(sub); deleteFd.push_back(fds[1]);//记录当前的写端供下个子进程用}在函数CreateSubProcesses内，先建立父进程相连的匿名管道，然后创建子进程，子进程也拷贝了一份父进程的文件描述符表，能通过文件描述符连接到匿名管道，因此父子进程通信的管道建立完成。在父进程语句中，需要注意的是，通过传参数子进程的pid和此时子进程读端对于的父进程的写端fd给SubEP类构建子进程对象，并且将对象放进数组subs里。在子进程的语句中，通过receiveTask函数获取任务码代码语言：c++复制
int receiveTask(int readfd){ int retcode=0;//返回任务码 ssize_t s= read(readfd,&retcode,sizeof(retcode));//从读端读出来的任务码放到retcode里 cout<<"process has read the TaskCode: "<<retcode<<endl; if(s==sizeof(int)) return retcode; else if(s<=0)return -1; else return 0;}让子进程在receiveTask函数中阻塞读取管道里的数据。前提已知父进程往匿名管道写入整数数据，数据范围为0,任务个数-1即任务数组对应的下标范围，子进程把读取到的数据存到变量retcode里，然后判断retcode是否是整数数据大小，如果是就返回数据给上层CreateSubProcesses函数，如果不是就返回-1。当变量taskcode接收到receiveTask函数返回的任务码时，如果任务码符合范围0,任务个数-1即父子进程按照我们的意愿通信正常，然后子进程拿着任务码调用funcMap数组执行任务；但如果接收的返回值是-1，则是父子进程间通信不正常，直接退出判断语句。
这里提到的子进程操作主要是子进程阻塞读取父进程写入的数据，还有子进程拿到数据执行任务。
对父进程操作代码语言：c++复制
void loadBalanceContrl(const vector& subs,const vector &funcMap,int comcode){ int processnum=subs.size();//子进程的个数 int tasknum=funcMap.size();//任务的个数 bool numoftime=(comcode==0?true:false);//若命令码是0则一直运行，若命令码为正数x，则允许x次后退出 while(true) { //rand()为伪随机数 //1.找到哪一个子进程 int subIndex=rand()%processnum; //2.找到哪一个执行哪一个任务 int taskIndex=rand()%tasknum; //3.任务发送给选择的进程 sendTask(subs[subIndex],taskIndex);//第一个参数传第几个子进程，第二个参数传第几个任务 sleep(1); if(!numoftime) { comcode--; if(comcode==0) break; } } //走到这里则是父进程给子进程通信完了，需要逐个关闭子进程读端对于的写端--倒退关解决bug for(size_t i=0;i<subs.size();i++) { close(subs[i]._writefd); cout<<"close process: [ "<<i<<" ]'s writeeop"<<endl; }}loadBalanceContrl函数需要main函数传入子进程数组subs，任务数组funcMap和命令码comcode。comcode用来指定父进程发送多少次数据给子进程即子进程需要执行多少次任务numoftime用来鉴别父进程需要写入多少次数据，当comcode为0时则numoftime为真，则父进程死循环往匿名管道里写数据；若命令码为正数x为非0，则numoftime为假，则父进程往匿名管道里写x次数据。通过sendTask函数让父进程选择指定的子进程，写入指定的任务码到匿名管道中代码语言：c++复制
void sendTask(const SubEp& process, int tasknum){ cout<<"send Task num: "<<tasknum<<" to the process: "<<process._name<<endl;//打印日志：任务几发送给几号子进程 ssize_t n=write(process._writefd,&tasknum,sizeof(tasknum));//该子进程读端对于的写端往管道里写入任务几-4个字节的数据 assert(n==sizeof(int));//判断写入的数据是否是4个字节 (void)n;}父子间进程通信完之后，按照子进程创建时间从先往后依次关闭子进程读端对应的父进程的写端。回收子进程代码语言：c++复制
void waitProcess(const vector& processes){ for(size_t i=0;i<processes.size();i++) { waitpid(processes[i]._subid,nullptr,0); cout<<"wait success for process: "<<processes[i]._subid<<endl; }}按照子进程创建时间从先往后依次回收子进程。整体代码代码语言：c++复制
#include#include#include#include#include#include#include#includeusing namespace std;#define PROCESS_NUM 3#define MakeSeed() srand((unsigned long)time(nullptr)^getpid()^rand()%1234)//建立伪随机数种子//创建父进程给子进程派发的任务列表typedef void(*func_t)();//函数指针类型，函数返回值为voidvoid downloadTask()//模拟下载任务{ cout<<getpid()<<": 下载任务n"<<endl; sleep(1);}void fflushTask()//模拟刷新任务{ cout<<getpid()<<": 刷新任务n"<<endl; sleep(1);}void subscribeTask()//模拟订阅任务{ cout<<getpid()<<": 订阅任务n"<<endl; sleep(1);}//把上面的三种任务load到列表中即让存放函数指针的vector的各个指针能够指向上面的函数，为了后面方便调用void loadTaskFunc(vector*out){ assert(out);//vector创建成功 out->push_back(downloadTask); out->push_back(fflushTask); out->push_back(subscribeTask);}//创建子进程对象class SubEp//endpoint---子进程对象{public:SubEp(pid_t subid,int writefd)//第一个参数是子进程的pid，第二个参数是该子进程读端对于父进程的写端fd:_subid(subid),_writefd(writefd){ char namebuffer[1024]; //第一个参数是表示第几号子进程，第二个参数是子进程的pid，第三个参数是该子进程读端对于的父进程的写端 snprintf(namebuffer,sizeof namebuffer,"process: %d [pid(%d) - fd(%d)]",num++,_subid,_writefd); _name=namebuffer;}public:static int num;string _name;pid_t _subid;int _writefd;//该子进程与父进程匿名管道对于的父进程的写端fd};int SubEp::num=0;int receiveTask(int readfd){ int retcode=0;//返回任务码 ssize_t s= read(readfd,&retcode,sizeof(retcode));//从读端读出来的任务码放到retcode里 cout<<"process has read the TaskCode: "<<retcode<<endl; if(s==sizeof(int)) return retcode; else if(s<=0)return -1; else return 0;}void CreateSubProcesses( vector*subs,vector& funcMap){ vector deleteFd;//创建子进程并且创建好父进程与各个子进程通信的管道int fds[2];for(size_t i=0;i<PROCESS_NUM;i++)//创建子进程{ int n=pipe(fds);//建立父子间进程的匿名管道--建立成功返回0，建立失败返回-1 assert(n==0);//判断管道是否建立成功 (void)n; pid_t id=fork();//创建子进程 if(id==0)//子进程进入判断语句 { for(size_t i=0;i=0 && taskcodepush_back(sub); deleteFd.push_back(fds[1]);//记录当前的写端供下个子进程用}}void sendTask(const SubEp& process, int tasknum){ cout<<"send Task num: "<<tasknum<<" to the process: "<<process._name<<endl;//打印日志：任务几发送给几号子进程 ssize_t n=write(process._writefd,&tasknum,sizeof(tasknum));//该子进程读端对于的写端往管道里写入任务几-4个字节的数据 assert(n==sizeof(int));//判断写入的数据是否是4个字节 (void)n;}void loadBalanceContrl(const vector& subs,const vector &funcMap,int comcode){ int processnum=subs.size();//子进程的个数 int tasknum=funcMap.size();//任务的个数 bool numoftime=(comcode==0?true:false);//若命令码是0则一直运行，若命令码为正数x，则允许x次后退出 while(true) { //rand()为伪随机数 //1.找到哪一个子进程 int subIndex=rand()%processnum; //2.找到哪一个执行哪一个任务 int taskIndex=rand()%tasknum; //3.任务发送给选择的进程 sendTask(subs[subIndex],taskIndex);//第一个参数传第几个子进程，第二个参数传第几个任务 sleep(1); if(!numoftime) { comcode--; if(comcode==0) break; } } //走到这里则是父进程给子进程通信完了，需要逐个关闭子进程读端对于的父进程写端 for(size_t i=0;i<subs.size();i++) { close(subs[i]._writefd); cout<<"close process: [ "<<i<<" ]'s writeeop"<<endl; // waitpid(subs[i]._subid,nullptr,0); // cout<<"wait success for process: "<<subs[i]._subid<<endl; }}void waitProcess(const vector& processes){ for(size_t i=0;i<processes.size();i++) { waitpid(processes[i]._subid,nullptr,0); cout<<"wait success for process: "<<processes[i]._subid<<endl; }}int main(){ MakeSeed();//建立伪随机数种子vector subs;//创建子进程对象并将子进程对象放进数组里vector funcMap;//建立一个任务表：父进程写入管道，子进程在管道读取，读取到的数据引导子进程去完成一些任务loadTaskFunc(&funcMap);//1.创建子进程并且创建好父进程与各个子进程通信的管道,并且让子进程阻塞等待父进程写入CreateSubProcesses(&subs,funcMap);//2.对父进程操作//父进程给子进程发送命令码，为0则一直运行，为正数x则运行x次后退出int Runcount=0;cout<>Runcount;cout<<endl;//这个函数负责让父进程给子进程发送命令码，让子进程去执行任务，要求子进程做到负载均衡 loadBalanceContrl(subs,funcMap,Runcount);//第一个参数是子进程列表，第二个参数任务列表，第三个参数是父进程给子进程发送的命令码 //3.回收子进程waitProcess(subs); return 0;}子进程具有读端未关闭的bug
image-20230522164129612
image-20230522164147020
image-20230522164456853通过上面的图例可以看到，2号子进程有一个写端与1号子进程的读端通信着。
得出结论：当父进程创建多个子进程，并且父进程作为写端而多个子进程作为读端从而进行进程间通信时，需要单独把子进程的所有写端都关闭。
这里提供两种方法关闭子进程的所有写端
方案一：在父进程创建子进程前构建一个vector对象，父进程创建子进程后，把父进程的写端放进vector里。等到父进程创建完下一个子进程时，vector里的写端即是当前子进程读端对应的上一个子进程的写端（有可能不只是一个写端），再把vector里的所有写端关闭即可。
image-20230522202611512
image-20230522203254968这次实现就是用的这个方案，其实不用也可以，因为当父进程往匿名管道里写完数据时，先把父进程对应各个子进程的写端全部关闭，然后再将全部子进程进行回收，这种顺序不会出现bug；但如果是按照创建子进程时间从旧往新关掉一个父进程的写端，然后立刻等待回收一个相应的子进程的话，会导致出现该子进程读端还有其他子进程的写端通信着，该子进程读端没有读到0导致子进程没有正常退出，那么父进程也就回收不到子进程。方案二：在父进程关闭写端，需要所有子进程关闭读端时，依次按照创建的时间从新往旧（从后往前）关闭父进程的写端。由于最后创建的子进程的读端只对应父进程的写端，那么父进程关闭写端时，最后一个子进程的读端读到0正常关闭读端，那么该子进程的文件描述符表也会被关闭，进而该子进程正常退出，从而该子进程连接着前一个子进程的写端也会被关闭；那么轮到下一个子进程时，该子进程的读端也只会对应父进程的写端，父进程关闭写端，子进程读端读到0正常关闭读端，子进程正常退出。
以上就是进程池设计的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/72356.html
微信扫一扫 
支付宝扫一扫 
