【Linux】使用管道实现一个简易版本的进程池

---

使用管道实现一个简易版本的进程池

流程图

---

代码

makefile

ProcessPool:ProcessPool.ccg++ -o $@ $^ -g -std=c++11
.PHONY:clean
clean:rm -f ProcessPool

---

Task.hpp

#pragma once#include <iostream>
#include <vector>typedef void (*task_t)(); //定义了一个函数指针类型task_t，它指向返回类型为void且不接受任何参数的函数。void task1()
{std::cout << "lol 刷新日志" << std::endl;
}
void task2()
{std::cout << "lol 更新野区，刷新出来野怪" << std::endl;
}
void task3()
{std::cout << "lol 检测软件是否更新,如果需要，就提示用户" << std::endl;
}
void task4()
{std::cout << "lol 用户释放技能，更新用的血量和蓝量" << std::endl;
}void LoadTask(std::vector<task_t> *tasks) // 该函数接受一个指向std::vector<task_t>的指针，并将其作为参数
{tasks->push_back(task1); //将task1函数的地址添加到向量中。tasks->push_back(task2);tasks->push_back(task3);tasks->push_back(task4);
}

---

ProcessPool.cc

#include "Task.hpp"  // 包含任务相关的头文件
#include <string>
#include <vector>
#include <cstdlib>
#include <ctime>
#include <cassert>
#include <unistd.h>
#include <sys/stat.h>
#include <sys/wait.h>const int processnum = 10;  // 设定进程池大小为10
std::vector<task_t> tasks; // 存储任务的向量// 定义channel类，用于管理进程间通信
class channel
{
public:channel(int cmdfd, int slaverid, const std::string &processname):_cmdfd(cmdfd), _slaverid(slaverid), _processname(processname){}
public:int _cmdfd;               // 用于向子进程发送命令的文件描述符pid_t _slaverid;          // 子进程IDstd::string _processname; // 子进程名称，用于日志显示
};// 子进程执行的函数
void slaver()
{while(true){int cmdcode = 0;// 从标准输入(被重定向到管道)读取命令int n = read(0, &cmdcode, sizeof(int)); if(n == sizeof(int)){std::cout <<"slaver say@ get a command: "<< getpid() << " : cmdcode: " <<  cmdcode << std::endl;// 执行对应的任务if(cmdcode >= 0 && cmdcode < tasks.size()) tasks[cmdcode]();}if(n == 0) break; // 管道关闭时退出}
}// 初始化进程池
void InitProcessPool(std::vector<channel> *channels)
{std::vector<int> oldfds;  // 存储历史文件描述符for(int i = 0; i < processnum; i++){int pipefd[2];int n = pipe(pipefd);  // 创建管道assert(!n);(void)n;pid_t id = fork();     // 创建子进程if(id == 0) // 子进程{// 关闭历史文件描述符std::cout << "child: " << getpid() << " close history fd: ";for(auto fd : oldfds) {std::cout << fd << " "; // 打印当前文件描述符的值，用于显示子进程正在关闭哪些文件描述符。close(fd); // 关闭文件描述符}std::cout << "\n";close(pipefd[1]);  // 关闭写端dup2(pipefd[0], 0);  // 将管道读端重定向到标准输入close(pipefd[0]); //关闭读端slaver();          // 执行子进程任务std::cout << "process : " << getpid() << " quit" << std::endl;exit(0);}// 父进程close(pipefd[0]);  // 关闭读端// 创建新的channel并添加到channels中std::string name = "process-" + std::to_string(i);channels->push_back(channel(pipefd[1], id, name));oldfds.push_back(pipefd[1]);sleep(1);}
}// 打印调试信息
void Debug(const std::vector<channel> &channels)
{for(const auto &c :channels){std::cout << c._cmdfd << " " << c._slaverid << " " << c._processname << std::endl;}
}// 显示菜单
void Menu()
{std::cout << "################################################" << std::endl;std::cout << "# 1. 刷新日志             2. 刷新出来野怪        #" << std::endl;std::cout << "# 3. 检测软件是否更新      4. 更新用的血量和蓝量  #" << std::endl;std::cout << "#                         0. 退出               #" << std::endl;std::cout << "#################################################" << std::endl;
}// 控制子进程执行任务
void ctrlSlaver(const std::vector<channel> &channels)
{int which = 0;while(true){int select = 0;Menu();std::cout << "Please Enter@ ";std::cin >> select;if(select <= 0 || select >= 5) break;int cmdcode = select - 1;// 轮询方式分配任务给子进程std::cout << "father say: " << " cmdcode: " <<cmdcode << " already sendto " << channels[which]._slaverid << " process name: " << channels[which]._processname << std::endl;write(channels[which]._cmdfd, &cmdcode, sizeof(cmdcode));which++;which %= channels.size();}
}// 清理进程池
void QuitProcess(const std::vector<channel> &channels)
{for(const auto &c : channels){close(c._cmdfd);  // 关闭所有管道waitpid(c._slaverid, nullptr, 0);  // 等待所有子进程结束}
}int main()
{LoadTask(&tasks);  // 加载任务列表srand(time(nullptr)^getpid()^1023);  // 初始化随机数种子std::vector<channel> channels; //InitProcessPool(&channels);  // 初始化进程池ctrlSlaver(channels);  // 控制子进程执行任务QuitProcess(channels);  // 清理进程池return 0;
}

---

程序流程：

1.main函数首先调用LoadTask(&tasks)，将task1到task4四个任务的函数地址存入全局tasks向量。

2.srand(time(nullptr)^getpid()^1023); 初始化随机数种子

3.std::vector<channel> channels;，这行代码的作用是定义一个名为 channels 的向量（std::vector），用于存储 channel 类型的对象。它的主要作用是管理多个 channel 对象，每个 channel 对象代表一个子进程的通信通道。

• 每个 channel 对象包含以下信息：

  • _cmdfd：用于向子进程发送命令的文件描述符（管道写端）。

  • _slaverid：子进程的进程ID（PID）。

  • _processname：子进程的名称，用于日志和调试。

• channels 向量存储了所有子进程的通信信息，父进程可以通过它管理所有子进程。

4.InitProcessPool(&channels); ，初始化进程池

// 初始化进程池
void InitProcessPool(std::vector<channel> *channels)
{std::vector<int> oldfds;  // 存储历史文件描述符for(int i = 0; i < processnum; i++){int pipefd[2];int n = pipe(pipefd);  // 创建管道assert(!n);(void)n;pid_t id = fork();     // 创建子进程if(id == 0) // 子进程{// 关闭历史文件描述符std::cout << "child: " << getpid() << " close history fd: ";for(auto fd : oldfds) {std::cout << fd << " ";close(fd);}std::cout << "\n";close(pipefd[1]);  // 关闭写端dup2(pipefd[0], 0);  // 将管道读端重定向到标准输入close(pipefd[0]); //关闭读端slaver();          // 执行子进程任务std::cout << "process : " << getpid() << " quit" << std::endl;exit(0);}// 父进程close(pipefd[0]);  // 关闭读端// 创建新的channel并添加到channels中std::string name = "process-" + std::to_string(i);channels->push_back(channel(pipefd[1], id, name));oldfds.push_back(pipefd[1]);sleep(1);}
}

5.std::vector<int> oldfds; 的作用是存储父进程中已经创建的管道的写端文件描述符（pipefd[1]）。它的主要目的是在创建新的子进程时，确保子进程能够关闭不需要的文件描述符，避免资源泄露和潜在的问题。

为什么需要 oldfds？

1. 文件描述符的继承：

   • 当父进程通过 fork() 创建子进程时，子进程会继承父进程的所有打开的文件描述符。

   • 如果父进程创建了多个管道（每个子进程对应一个管道），那么每个子进程都会继承所有管道的文件描述符，即使这些管道是用于其他子进程的。

2. 资源泄露问题：

   • 如果子进程不关闭不需要的文件描述符，这些文件描述符会一直保持打开状态，导致资源泄露。

   • 例如，假设父进程创建了 10 个子进程，每个子进程都会继承 10 个管道的文件描述符，但实际上每个子进程只需要一个管道的读端文件描述符。

3. 避免干扰：

   • 如果子进程不关闭不需要的文件描述符，可能会导致意外的行为。例如，某个子进程可能会错误地读取其他子进程的管道数据。

6.for(int i = 0; i < processnum; i++)，循环 processnum=10 次，每次创建一个子进程和一个管道。

7.int pipefd[2];

pipefd 是一个长度为 2 的整型数组，用于存储管道的两个文件描述符：

• pipefd[0]：管道的 读端文件描述符，用于从管道中读取数据。
• pipefd[1]：管道的 写端文件描述符，用于向管道中写入数据。

8.int n = pipe(pipefd);

调用 pipe 系统函数来创建一个管道，并将结果存储在变量 n 中。

1. pipe 系统函数的作用

pipe 是一个系统调用，用于创建一个管道。管道的本质是一个内核缓冲区，用于在两个进程之间传递数据。管道有两个端点：

• 读端：用于从管道中读取数据。
• 写端：用于向管道中写入数据。

pipe 函数的原型如下：

int pipe(int pipefd[2]);

---

2. 参数 pipefd[2]

• pipefd 是一个长度为 2 的整型数组，用于存储管道的两个文件描述符：
  • pipefd[0]：管道的 读端文件描述符，用于从管道中读取数据。
  • pipefd[1]：管道的 写端文件描述符，用于向管道中写入数据。

---

3. 返回值 n

• 如果 pipe 调用成功，返回 0。
• 如果 pipe 调用失败，返回 -1，并设置 errno 表示错误原因。

---

4. 代码解析

int n = pipe(pipefd);

• pipe(pipefd)：调用 pipe 函数创建管道。
• n：存储 pipe 函数的返回值，用于检查管道是否创建成功。

9.assert(!n);，(void)n;

• assert(!n)：确保管道创建成功。如果 pipe 调用失败，程序会终止。
• (void)n：忽略未使用的变量警告。

10.pid_t id = fork(); ，创建子进程

if(id == 0) // 子进程
{// 关闭历史文件描述符std::cout << "child: " << getpid() << " close history fd: ";for(auto fd : oldfds) {std::cout << fd << " ";close(fd);}std::cout << "\n";close(pipefd[1]);  // 关闭写端dup2(pipefd[0], 0);  // 将管道读端重定向到标准输入close(pipefd[0]); //关闭读端slaver();          // 执行子进程任务std::cout << "process : " << getpid() << " quit" << std::endl;exit(0);
}

11.在子进程中，id 为 0。

12.std::cout << "child: " << getpid() << " close history fd: ";

打印当前子进程的PID，用于区分不同子进程

" close history fd: "，说明接下来要关闭的文件描述符

for(auto fd : oldfds) {std::cout << fd << " ";// 打印当前文件描述符的值，用于显示子进程正在关闭哪些文件描述符。close(fd);// 关闭文件描述符
}

在子进程中遍历 oldfds 向量，关闭所有不需要的文件描述符。

具体来说，它的目的是确保子进程只保留与自己相关的文件描述符，关闭其他无关的文件描述符，从而避免资源泄露和潜在的问题。

close(pipefd[1]);  // 子进程关闭写端，因为子进程只需要读取命令
dup2(pipefd[0], 0);  // 将父进程管道读端重定向到标准输入
close(pipefd[0]); //关闭父进程读端
slaver();          // 执行子进程任务

dup2函数将管道的读端(pipefd[0])复制到标准输入(0)

这意味着之后从标准输入读取的数据实际上是从管道读取的

后续代码中可以直接使用read(0,…)来读取父进程发送的数据

数据流向：父进程 ---> 写端(pipefd[1]) ---> 管道 ---> 读端(重定向到标准输入) ---> 子进程

子进程：

• 关闭写端(pipefd[1])
• 将读端重定向到标准输入
• 关闭原读端(因为已重定向)

15.进入子进程函数

// 子进程执行的函数
void slaver()
{while(true){int cmdcode = 0;// 从标准输入(被重定向到管道)读取命令int n = read(0, &cmdcode, sizeof(int)); if(n == sizeof(int)){std::cout <<"slaver say@ get a command: "<< getpid() << " : cmdcode: " <<  cmdcode << std::endl;// 执行对应的任务if(cmdcode >= 0 && cmdcode < tasks.size()) tasks[cmdcode]();}if(n == 0) break; // 管道关闭时退出}
}

while(true)，无限循环，持续监听命令

int cmdcode = 0;
int n = read(0, &cmdcode, sizeof(int)); 

read(0, …)：从标准输入读取数据，因为前面做了重定向，实际是从管道读取

&cmdcode：存储读取数据的地址

sizeof(int)：读取int大小的数据

n：返回实际读取的字节数

if(n == sizeof(int)) {  // 成功读取到完整的命令// 打印调试信息std::cout <<"slaver say@ get a command: "<< getpid() << " : cmdcode: " <<  cmdcode << std::endl;// 执行对应任务if(cmdcode >= 0 && cmdcode < tasks.size()) tasks[cmdcode]();  // 调用任务函数
}

if(cmdcode >= 0 && cmdcode < tasks.size())，确保cmdcode非负，确保cmdcode小于任务数组大小，防止数组越界访问

tasks[cmdcode]();，tasks[cmdcode]获取对应的函数指针，()操作符调用该函数。

// 假设cmdcode = 0
tasks[0](); // 调用task1()，输出"lol 刷新日志"// 假设cmdcode = 1
tasks[1](); // 调用task2()，输出"lol 更新野区，刷新出来野怪"// 假设cmdcode = 2
tasks[2](); // 调用task3()，输出"lol 检测软件是否更新"// 假设cmdcode = 3
tasks[3](); // 调用task4()，输出"lol 更新用户血量和蓝量"

if(n == 0) break; ，管道关闭时退出

16.slaver()结束，返回刚刚的

std::cout << "process : " << getpid() << " quit" << std::endl; //打印退出信息,getpid帮助我们确认哪个进程正在退出
exit(0); // 立即终止当前进程

17.然后执行InitProcessPool()函数的剩下来部分

// 父进程
close(pipefd[0]);  // 关闭读端// 创建新的channel并添加到channels中
std::string name = "process-" + std::to_string(i);
channels->push_back(channel(pipefd[1], id, name));
oldfds.push_back(pipefd[1]);sleep(1);

close(pipefd[0]);，父进程只需写入命令，不需要读。及时关闭不需要的文件描述符

std::string name = "process-" + std::to_string(i);，为每个子进程创建唯一名称。

std::to_string(i) : 将数字i转为字符串，“+” : 字符串拼接运算符。

效果如：process-0, process-1, process-2…

channels->push_back(channel(pipefd[1], id, name));，push_back在容器末尾添加新元素。创建临时 channel 对象并添加到 vector

channel是一个结构体，存储子进程信息：

void InitProcessPool(std::vector<channel> *channels)struct channel {int fd;      // 管道写端pid_t pid;   // 子进程IDstd::string name;  // 进程名称channel(int _fd, pid_t _pid, const std::string& _name): fd(_fd), pid(_pid), name(_name){}
};

oldfds.push_back(pipefd[1]);，添加管道写端的文件描述符。

保存文件描述符的用途：

• 用于后续关闭文件描述符
• 防止文件描述符泄漏
• 进程间通信的管理
• 资源清理

sleep(1);，休眠1s。

18.进入main函数，执行ctrlSlaver(channels);

// 控制子进程执行任务
void ctrlSlaver(const std::vector<channel> &channels)
{int which = 0;while(true){int select = 0;Menu();std::cout << "Please Enter@ ";std::cin >> select;if(select <= 0 || select >= 5) break;int cmdcode = select - 1;// 轮询方式分配任务给子进程std::cout << "father say: " << " cmdcode: " <<cmdcode << " already sendto " << channels[which]._slaverid << " process name: " << channels[which]._processname << std::endl;write(channels[which]._cmdfd, &cmdcode, sizeof(cmdcode));which++;which %= channels.size();}
}

轮询机制

int which = 0;  // 轮询索引
which++;
which %= channels.size();  // 循环轮询

实现了循环分配任务给不同子进程

如果有3个进程，which的值会是 0,1,2,0,1,2…

---

任务选择

while(true) {int select = 0;Menu();  // 显示菜单std::cout << "Please Enter@ ";std::cin >> select;  // 获取用户输入if(select <= 0 || select >= 5) break;  // 退出条件int cmdcode = select - 1;  // 将用户输入的选项编号转换为程序内部使用的命令代码。
}

---

发送任务示例

// 显示任务分配信息
std::cout << "father say: " << " cmdcode: " << cmdcode << " already sendto " << channels[which]._slaverid << " process name: " << channels[which]._processname << std::endl;// 向子进程发送命令
write(channels[which]._cmdfd, &cmdcode, sizeof(cmdcode));

cmdcode要执行的命令编号（0代表hello，1代表calc等）

_slaverid: 子进程的PID（进程ID）

_processname: 子进程的名称

write(channels[which]._cmdfd, &cmdcode, sizeof(cmdcode));

channels[which]._cmdfd：管道的写端文件描述符

&cmdcode：命令代码的地址

sizeof(cmdcode)：发送的字节数（int类型通常是4字节）

19.返回主函数，执行QuitProcess(channels);，清理进程池。

void QuitProcess(const std::vector<channel> &channels)
{// 遍历所有channel对象for(const auto &c : channels){// 1. 关闭管道close(c._cmdfd);  // 关闭管道写端// 2. 等待子进程结束waitpid(c._slaverid, nullptr, 0);  // 阻塞等待直到子进程结束}
}

20.return 0；