匿名管道--任务派发程序

整体代码逻辑架构图如下整体架构图如下整体代码如下:main.cc:#include processpool.hpp #include Tasks.hpp int main(int argc, char *argv[]) { if(argc!2){ coutUsage: argv[0] process_numendl; return -1; } int process_num stoi(argv[1]); proc_pool *pool new proc_pool(process_num, worker); //初始化进程池 pool-Initproc_pool(); //分发任务 pool-Dispatch_task(); sleep(3); //打印进城池 pool-Print(); //清空进程池 pool-Delete_proc_pool(); delete pool; return 0; }processpool.hpp:#ifndef PROCESSPOOL_HPP #define PROCESSPOOL_HPP #include iostream #include vector #include functional #include unistd.h #include sys/types.h #include sys/wait.h #include string #include Tasks.hpp using namespace std; using work_t functionvoid(); // 定义函数指针类型 enum : int { OK, FORK_ERROR, PIPE_ERROR, READ_ERROR, WRITE_ERROR, EXIT_ERROR }; // 先描述 class channel { public: channel(int procfd, pid_t who) : _procfd(procfd), who(who) {} // 构造函数 ~channel() {} string Name() { return name channel_ to_string(who) - to_string(_procfd); // 获取子进程名 } // 4个get函数 void send(int cmd) { ::write(_procfd, cmd, sizeof(cmd)); // 发送命令 } bool Close() { if (_procfd 0) return true; if (::close(_procfd) -1) { perror(close); return false; } _procfd -1; // 文件描述符必定大于0,因此-1无效 return true; } pid_t Id() { return who; // 获取子进程id } int procid() { return _procfd; // 获取管道fd } private: int _procfd; string name; pid_t who; }; // 再组织 class proc_pool { public: proc_pool(int n, work_t work) : _work(work), proc_num(n) { } ~proc_pool() { } int Initproc_pool() { // 创建n个子进程 for (int i 0; i proc_num; i) { // 1.创建管道 int pipefd[2]; int n pipe(pipefd); if (n 0) { return PIPE_ERROR; // 管道创建失败 } // 2.创建父子进程 pid_t pid fork(); if (pid 0) { return FORK_ERROR; // 创建进程失败 } if (pid 0) { // 子进程关闭历史fd for (auto ch : _channels) { cout 已关闭 _procfd ch.procid() endl; ch.Close(); } // 子进程关闭写端 ::close(pipefd[1]); cout 子进程获取到读端fd: pipefd[0] endl; // 子进程重定向标准输入为读端 dup2(pipefd[0], 0); // 子进程开始完成父进程分发的随机任务 worker(); // 子进程退出 _exit(OK); // 子进程退出 } // 3.父进程关闭读端子进程关闭写端 ::close(pipefd[0]); // 4.父进程保存子进程id和管道fd _channels.emplace_back(pipefd[1], pid); } return OK; } void Dispatch_task() { int who 0; int num 20; while (num--) { // 1.选择一个任务 int taskcode tasks.Task_select(); // 2.选择一个子进程管道 // 先给管道内部成员起别名方便操作 channel ch _channels[who % proc_num]; sleep(1); cout 父进程分发任务码 taskcode ch.Name() endl; cout 任务还剩: num 个 endl; // 3.发送任务码给子进程 ch.send(taskcode); } } void Delete_proc_pool() { // 删除管道 for (auto ch : _channels) { bool retch.Close();//close后,写端关闭,子进程读取到EOF,退出循环,退出子进程 pid_t pid waitpid(ch.Id(), nullptr, 0); // 等待子进程退出 if(rettruepid0){ cout 子进程退出成功ID pid endl; } else{ cout 子进程退出失败ID pid endl; } } } void Print(){ for (auto ch : _channels) { cout 进程池元素正在循环打印中 ch.Name() endl; } } private: vectorchannel _channels; work_t _work; int proc_num; }; #endifTasks.hpp:#ifndef TASKS_HPP #define TASKS_HPP #include iostream #include vector #include functional #include unistd.h #include sys/types.h #include ctime using namespace std; using task_tfunctionvoid(); // 定义函数指针类型 class Tasks { public: Tasks() { srand(time(NULL)); task_queue.push_back([](){coutgetpid() 进程任务1endl;}); task_queue.push_back([](){coutgetpid() 进程任务2endl;}); task_queue.push_back([](){coutgetpid() 进程任务3endl;}); task_queue.push_back([](){coutgetpid() 进程任务4endl;}); } ~Tasks() {} //随机选择任务码 int Task_select(){ return rand()%task_queue.size(); } //执行vector函数队列中的任务 void Execute_task(unsigned long int num){ if(num task_queue.size()||num0) { task_queue[num](); } else { cout任务码不存在endl; } } private: vectortask_t task_queue; }; //任务对象 Tasks tasks; //子进程进行完成任务 void worker(){ while (true) { int cmd0; // 从管道读取任务码,直接拿pipefd是拿不到的所以之前dup2重定向了标准输入所以这里直接从标准输入读取 int n read(0, cmd, sizeof(cmd)); if(n 0) { cout子进程: getpid()读取到文件EOFendl; break; // 读取失败退出循环 } else { tasks.Execute_task(cmd); } } } #endifMakefile:SRC$(wildcard *.cc) OBJ$(SRC:.cc.o) BINpipe $(BIN):$(OBJ) g -o $ $^ %.o:%.cc g -c $ .PHONY:clean clean: rm -rf $(BIN) $(OBJ) .PHONY:test test: echo $(SRC) echo $(OBJ)