进程池（C/C++）

C语言实现/* * 进程池示例 * 使用消息队列进行任务分发 */#includestdio.h#includestdlib.h#includeunistd.h#includesys/wait.h#includesys/msg.h#includestring.h#defineWORKER_NUM3// 进程池中工作进程的数量#defineMSG_KEY12345// 消息队列的 key#defineMSG_TYPE1// 消息类型/* 消息结构体必须包含 long type */structmsgbuf{longtype;// 消息类型用于区分inttask_id;// 任务编号};/* 子进程工作进程入口函数 */voidworker_process(intmsgid){structmsgbufmsg;while(1){/* * 从消息队列接收任务 * msg.type MSG_TYPE只接收该类型的消息 * 0阻塞等待 */msgrcv(msgid,msg,sizeof(msg.task_id),MSG_TYPE,0);printf([Worker PID%d] start task %d\n,getpid(),msg.task_id);sleep(2);// 模拟任务执行耗时printf([Worker PID%d] finish task %d\n,getpid(),msg.task_id);}}intmain(){pid_tpid;intmsgid;/* 创建消息队列用于父进程给子进程发任务 */msgidmsgget(MSG_KEY,IPC_CREAT|0666);if(msgid0){perror(msgget);exit(1);}/* 创建进程池 */for(inti0;iWORKER_NUM;i){pidfork();if(pid0){perror(fork);exit(1);}if(pid0){// 子进程成为 workerworker_process(msgid);exit(0);// worker 不会回到 main}else{// 父进程继续创建下一个 workerprintf(Create worker PID%d\n,pid);}}/* 父进程发送任务 */for(inttask_id1;task_id10;task_id){structmsgbufmsg;msg.typeMSG_TYPE;msg.task_idtask_id;/* * 发送任务到消息队列 * 子进程会通过 msgrcv 竞争获取任务 */msgsnd(msgid,msg,sizeof(msg.task_id),0);printf([Master] send task %d\n,task_id);sleep(1);// 控制任务产生速度}/* 回收子进程实际工程中通常会用信号优雅退出 */while(1){wait(NULL);}return0;} 编译 运行gcc process_pool.c -o process_pool./process_pool**c实现**#includeiostream#includevector#includeunistd.h#includesys/wait.h#includefcntl.h#includecstring#includealgorithm/* * Worker 进程类 * 每个 Worker 绑定一条管道 */classWorker{public:pid_t pid;// 子进程 PIDintpipe_fd;// 读端文件描述符Worker(pid_t p,intfd):pid(p),pipe_fd(fd){}};/* * 进程池类 */classProcessPool{private:std::vectorWorkerworkers;size_t next_worker0;// 轮询索引RR 调度public:/* * 构造函数创建进程池 */ProcessPool(size_t worker_num){for(size_t i0;iworker_num;i){intpipefd[2];// 创建管道if(pipe(pipefd)-1){perror(pipe);exit(1);}pid_t pidfork();if(pid0){perror(fork);exit(1);}if(pid0){// 子进程 close(pipefd[1]);// 关闭写端worker_loop(pipefd[0]);exit(0);}else{// 父进程 close(pipefd[0]);// 关闭读端workers.emplace_back(pid,pipefd[1]);std::cout[Master] Create worker pidpidstd::endl;}}}/* * 子进程主循环 */voidworker_loop(intread_fd){inttask_id;while(true){ssize_t nread(read_fd,task_id,sizeof(task_id));if(n0)break;// 管道关闭std::cout[Worker PIDgetpid()] start task task_idstd::endl;sleep(2);// 模拟任务执行std::cout[Worker PIDgetpid()] finish task task_idstd::endl;}}/* * 分发任务Round Robin */voiddispatch(inttask_id){size_t idxnext_worker;next_worker(next_worker1)%workers.size();write(workers[idx].pipe_fd,task_id,sizeof(task_id));std::cout[Master] assign task task_id to worker workers[idx].pidstd::endl;}/* * 回收子进程 */~ProcessPool(){for(autow:workers){close(w.pipe_fd);waitpid(w.pid,nullptr,0);}}};intmain(){ProcessPoolpool(3);// 创建 3 个 workerfor(inti1;i10;i){pool.dispatch(i);sleep(1);// 控制任务速率}return0;}