整体代码逻辑架构图如下：
![]()
整体架构图如下：
![]()
![]()
整体代码如下:
main.cc:
#include "processpool.hpp" #include "Tasks.hpp" int main(int argc, char *argv[]) { if(argc!=2){ cout<<"Usage: "<<argv[0]<<" <process_num>"<<endl; 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 = function<void()>; // 定义函数指针类型 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 ret=ch.Close();//close后,写端关闭,子进程读取到EOF,退出循环,退出子进程 pid_t pid = waitpid(ch.Id(), nullptr, 0); // 等待子进程退出 if(ret==true&&pid>0){ cout << "子进程退出成功，ID=" << pid << endl; } else{ cout << "子进程退出失败，ID=" << pid << endl; } } } void Print(){ for (auto &ch : _channels) { cout << "进程池元素正在循环打印中：" << ch.Name() << endl; } } private: vector<channel> _channels; work_t _work; int proc_num; }; #endif
Tasks.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_t=function<void()>; // 定义函数指针类型 class Tasks { public: Tasks() { srand(time(NULL)); task_queue.push_back([](){cout<<getpid()<<" ：进程"<<"任务1"<<endl;}); task_queue.push_back([](){cout<<getpid()<<" ：进程"<<"任务2"<<endl;}); task_queue.push_back([](){cout<<getpid()<<" ：进程"<<"任务3"<<endl;}); task_queue.push_back([](){cout<<getpid()<<" ：进程"<<"任务4"<<endl;}); } ~Tasks() {} //随机选择任务码 int Task_select(){ return rand()%task_queue.size(); } //执行vector函数队列中的任务 void Execute_task(unsigned long int num){ if(num < task_queue.size()||num<0) { task_queue[num](); } else { cout<<"任务码不存在"<<endl; } } private: vector<task_t> task_queue; }; //任务对象 Tasks tasks; //子进程进行完成任务 void worker(){ while (true) { int cmd=0; // 从管道读取任务码,直接拿pipefd是拿不到的，所以之前dup2重定向了标准输入，所以这里直接从标准输入读取 int n= read(0, &cmd, sizeof(cmd)); if(n == 0) { cout<<"子进程: "<<getpid()<<"读取到文件EOF"<<endl; break; // 读取失败，退出循环 } else { tasks.Execute_task(cmd); } } } #endif
Makefile:
SRC=$(wildcard *.cc) OBJ=$(SRC:.cc=.o) BIN=pipe $(BIN):$(OBJ) g++ -o $@ $^ %.o:%.cc g++ -c $< .PHONY:clean clean: rm -rf $(BIN) $(OBJ) .PHONY:test test: @echo $(SRC) @echo $(OBJ)
