当前位置：首页 > news >正文

进程池（C/C++）

news 2026/5/15 1:15:25

C语言实现

/* * 进程池示例 * 使用消息队列进行任务分发 */#include<stdio.h>#include<stdlib.h>#include<unistd.h>#include<sys/wait.h>#include<sys/msg.h>#include<string.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;/* 创建消息队列（用于父进程给子进程发任务） */msgid=msgget(MSG_KEY,IPC_CREAT|0666);if(msgid<0){perror("msgget");exit(1);}/* 创建进程池 */for(inti=0;i<WORKER_NUM;i++){pid=fork();if(pid<0){perror("fork");exit(1);}if(pid==0){// 子进程：成为 workerworker_process(msgid);exit(0);// worker 不会回到 main}else{// 父进程：继续创建下一个 workerprintf("Create worker PID=%d\n",pid);}}/* 父进程：发送任务 */for(inttask_id=1;task_id<=10;task_id++){structmsgbufmsg;msg.type=MSG_TYPE;msg.task_id=task_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++实现

#include<iostream>#include<vector>#include<unistd.h>#include<sys/wait.h>#include<fcntl.h>#include<cstring>#include<algorithm>/* * Worker 进程类 * 每个 Worker 绑定一条管道 */classWorker{public:pid_t pid;// 子进程 PIDintpipe_fd;// 读端文件描述符Worker(pid_t p,intfd):pid(p),pipe_fd(fd){}};/* * 进程池类 */classProcessPool{private:std::vector<Worker>workers;size_t next_worker=0;// 轮询索引（RR 调度）public:/* * 构造函数：创建进程池 */ProcessPool(size_t worker_num){for(size_t i=0;i<worker_num;++i){intpipefd[2];// 创建管道if(pipe(pipefd)==-1){perror("pipe");exit(1);}pid_t pid=fork();if(pid<0){perror("fork");exit(1);}if(pid==0){// ===== 子进程 =====close(pipefd[1]);// 关闭写端worker_loop(pipefd[0]);exit(0);}else{// ===== 父进程 =====close(pipefd[0]);// 关闭读端workers.emplace_back(pid,pipefd[1]);std::cout<<"[Master] Create worker pid="<<pid<<std::endl;}}}/* * 子进程主循环 */voidworker_loop(intread_fd){inttask_id;while(true){ssize_t n=read(read_fd,&task_id,sizeof(task_id));if(n<=0)break;// 管道关闭std::cout<<"[Worker PID="<<getpid()<<"] start task "<<task_id<<std::endl;sleep(2);// 模拟任务执行std::cout<<"[Worker PID="<<getpid()<<"] finish task "<<task_id<<std::endl;}}/* * 分发任务（Round Robin） */voiddispatch(inttask_id){size_t idx=next_worker;next_worker=(next_worker+1)%workers.size();write(workers[idx].pipe_fd,&task_id,sizeof(task_id));std::cout<<"[Master] assign task "<<task_id<<" to worker "<<workers[idx].pid<<std::endl;}/* * 回收子进程 */~ProcessPool(){for(auto&w:workers){close(w.pipe_fd);waitpid(w.pid,nullptr,0);}}};intmain(){ProcessPoolpool(3);// 创建 3 个 workerfor(inti=1;i<=10;++i){pool.dispatch(i);sleep(1);// 控制任务速率}return0;}