c++进程池（Linux）的实现（2025.12.22）
学习比特课程后，学习总结和代码实现。
这节课产生了两点困惑，查阅资料后，有一下理解：
1.“FD 数值相同”≠“指向同一个管道”
比如两次pipe()可能都生成pipefd[0]=3（因为前一轮父进程关闭了读端 3，FD 号被复用），但内核中会通过 “进程 ID + FD 号” 来区分
2.系统调用read()函数，什么时候阻塞，什么时候不阻塞呢？
写端未关闭时候：阻塞等待，直到有数据写入（哪怕只写 1 字节），读取后返回实际字节数
写端关闭的时候：管道空时read()立即返回 0（EOF），且后续再调用read()也一直返回 0
除了上面易混淆的点，再补充一点我对进程池的理解：
当任务来时开启进程，任务结束时关闭进程，这样会产生多次系统调用，给系统带来更多负载，而进程池可以先创建多个进程处于“随时待命”的状态，减少多次“开启进程”“关闭进程”的动作，减少操作系统的工作量。
接下来是代码总结：
.
├── Main.cc //测试
├── Makefile
├── ProcessPool.hpp //进程池代码
└── Task.hpp //任务代码
//ProcessPool.hpp #ifndef __PROCESS_POOL__ #define __PROCESS_POOL__ #include <iostream> #include <cstdlib> #include <vector> #include <unistd.h> #include "Task.hpp" #include <sys/wait.h> //管道封装 class Channel { public: Channel(int fd, pid_t id) : _wfd(fd), _subid(id) { _name = "channel-" + std::to_string(_wfd) + "-" + std::to_string(_subid); }; ~Channel() {}; void Send(int code) { int n = write(_wfd, &code, sizeof(code)); (void)n; } void Close() { close(_wfd); } void Wait() { pid_t rid = waitpid(_wfd, nullptr, 0); } int FD() { return _wfd; } pid_t SubId() { return _subid; } std::string Name() { return _name; } private: int _wfd; pid_t _subid; std::string _name; // int nodenum }; class ChannelManager { public: ChannelManager() : _next(0) {}; void Insert(int wfd, pid_t subid) { _channels.emplace_back(wfd, subid); } Channel &Select() { auto &c = _channels[_next]; _next++; _next %= _channels.size(); return c; } void PrintChannel() { for (auto &channel : _channels) { std::cout << channel.Name() << std::endl; } } void StopSubProcess() { for (auto &channel : _channels) { channel.Close(); std::cout << "关闭:" << channel.Name() << std::endl; } } void WaitSubProsess() { for (auto &channel : _channels) { channel.Wait(); std::cout << "回收子进程:" << channel.Name() << std::endl; } } ~ChannelManager() {}; private: std::vector<Channel> _channels; int _next; }; const int gdefaultnum = 5; //默认开启进程数，可改 //进程池 class ProcessPool { public: ProcessPool(int num) : _process_num(num) { _tm.Register(PrintLog); _tm.Register(Download); _tm.Register(Upload); } void Work(int rfd) { while (true) { int code = 0; ssize_t n = read(rfd, &code, sizeof(code)); if (n > 0) { if (n != sizeof(code)) { continue; } std::cout << "子进程[" << getpid() << "]收到到一个任务码:" << code << std::endl; _tm.Execute(code); } else if (n == 0) { std::cout << "子进程退出\n"; break; } else { std::cout << "读取错误\n"; break; } } } bool Create() { for (int i = 0; i < _process_num; i++) { // 1.创建管道 int pipefd[2] = {0}; int n = pipe(pipefd); if (n < 0) { return false; } // 2.创建子进程 pid_t subid = fork(); if (subid < 0) return false; else if (subid == 0) { // 子进程 // 3.关闭不需要的文件描述符 close(pipefd[1]); Work(pipefd[0]); close(pipefd[0]); exit(0); } else { // 父进程 // 3.关闭不需要的文件描述符 close(pipefd[0]); _cm.Insert(pipefd[1], subid); } } return true; } void Debug() { _cm.PrintChannel(); } void Run() { // 1.选择一个任务 int taskcode = _tm.Code(); // 2.选择一个信道[子进程]，负载均衡的选择一个子进程，完成任务 auto &c = _cm.Select(); std::cout << "选择子进程：" << c.Name() << std::endl; // 3.发送任务 c.Send(taskcode); std::cout << "发送一个任务码：" << taskcode << std::endl; } void Stop() { // 关闭父进程所有的wfd即可 _cm.StopSubProcess(); // 回收所有子进程 _cm.WaitSubProsess(); } ~ProcessPool() {}; private: ChannelManager _cm; int _process_num; TaskManager _tm; }; #endif//Task.hpp #pragma once #include <iostream> #include <vector> typedef void (*task_t)(); void PrintLog(){ std::cout << "我是一个打印日志任务" << std::endl; } void Download() { std::cout << "我是一个下载的任务" << std::endl; } void Upload() { std::cout << "我是一个上传的任务" << std::endl; } //////////////////////// class TaskManager { public: TaskManager() { srand(time(nullptr)); } void Register(task_t t) { _tasks.push_back(t); } int Code() { return rand() % _tasks.size(); } void Execute(int code) { if (code >= 0 && code < _tasks.size()) { _tasks[code](); } } ~TaskManager() {} private: std::vector<task_t> _tasks; };//Main.cc #include"ProcessPool.hpp" int main(){ ProcessPool pp(gdefaultnum); pp.Create(); // pp.Debug(); int cnt = 10; while (cnt--) { pp.Run(); sleep(1); } pp.Stop(); return 0; }