小学学校网站建设计划网络营销品牌
目录
2、进程池
1）理解进程池
2）进程池的实现
整体框架：
a. 加载任务
b. 先描述，再组织
I. 先描述
II. 再组织
c. 创建信道和子进程
d. 通过channel控制子进程
e. 回收管道和子进程
问题1：
解答1：
问题2：
解答2：
f. 将进程池本身和任务文件本身进行解耦
3）完整代码
processpool.cc：
Task.hpp：
Makefile：
命令行中的 | ，就是匿名管道
它们的父进程都是bash
2、进程池
1）理解进程池
a. 可以将任务写入管道来给到子进程，从而可以提前创建子进程想让哪个子进程完成任务，我就让写入到哪个子进程相对的管道中
b. 管道里面没有数据，worker进程就在阻塞等待，等待务的到来！！
c. master向哪一个管道进行写入，就是唤醒哪一个子进程来处理任务
d. 均衡的向后端子进程划分任务，称之为负载均衡父进程要进行后端任务的负载均衡
父进程直接向管道里写入固定长度的四字节(int)数组下标(任务码)
函数指针数组中元素分别指向不同的任务，以便master控制worker完成指定工作
2）进程池的实现
整体框架：
// ./processpool 3
int main(int argc, char *argv[])
{if (argc != 2){std::cerr << "Usage: " << argv[0] << " processnum" << std::endl;}int num = std::stoi(argv[1]);LoadTask(); // 加载任务std::vector<Channel> channels; // 将管道组织起来// 1.创建信道和子进程CreateChannelAndSub(num, &channels);// 2.通过channel控制子进程CtrlProcess(channels, 10);// 3.回收管道和子进程 a.关闭所有的写端 b.回收子进程ClearUpChannel(channels);return 0;
}a. 加载任务
我这里用的是打印的方式来模拟任务的分配，通过打印从而了解子进程执行任务的情况，通过种下随机数种子，产生随机数，进而随机的向子进程分配任务，work即为子进程需要做的工作
Task.hpp：
#pragma once#include <iostream> #include <ctime> #include <cstdlib> #include <sys/types.h> #include <unistd.h>#define TaskNum 3typedef void (*task_t)(); // task_t 函数指针void Print() {std::cout << "I am print task" << std::endl; } void DownLoad() {std::cout << "I am a download task" << std::endl; } void Flush() {std::cout << "I am a flush task" << std::endl; }task_t tasks[TaskNum];void LoadTask() {srand(time(nullptr) ^ getpid() ^ 177); // 种一个随机种子tasks[0] = Print;tasks[1] = DownLoad;tasks[2] = Flush; }void ExcuteTask(int number) {if(number < 0 || number > 2) return;tasks[number](); }int SelectTask() {return rand() % TaskNum; }void work(int rfd) {while (true){int command = 0;int n = read(rfd, &command, sizeof(command));if (n == sizeof(int)){std::cout << "pid is: " << getpid() << " handler task" << std::endl;ExcuteTask(command);}else if (n == 0){std::cout << "sub process: " << getpid() << " quit" << std::endl;break;}} }
// 命令行规范 --> ./processpool 3
int main(int argc, char *argv[])
{if (argc != 2){std::cerr << "Usage: " << argv[0] << " processnum" << std::endl;}int num = std::stoi(argv[1]);LoadTask(); // 加载任务return 0;
}b. 先描述，再组织
I. 先描述
需要控制的信道（即发送端wfd）数量多且繁琐，需要管理起来从而方便控制给子进程发送任务
class Channel { private:int _wfd;int _subprocessid;std::string _name; };在信道中，我们需要知道发送的文件描述符wfd，还有知道子进程的pid，以及信道的命名（用来区分信道）
II. 再组织
std::vector<Channel> channels;我们通过用一个vector数组将所有的Channel存储起来，从而实现对它们的增删查改，以方便管理
c. 创建信道和子进程
void CreateChannelAndSub(int num, std::vector<Channel> *channels)
{for (int i = 0; i < num; ++i){// 1.创建管道int pipefd[2] = {0};int n = pipe(pipefd);if (n < 0) exit(1); // 创建管道失败// 2.创建子进程pid_t id = fork();if (id == 0){// Childclose(pipefd[1]);work();close(pipefd[0]);exit(0);}// 3.构建一个名字std::string channel_name = "Channel-" + std::to_string(i);// Fatherclose(pipefd[0]);// 拿到了 a.子进程的pid b.父进程需要的管道的w端channels->push_back(Channel(pipefd[1], id, channel_name));}
}用命令行参数的方式传入得到的argv[1]即为输入命令需要的子进程和管道个数
通过for循环，创建 num个 pipe管道以及子进程，当创建完子进程时，需要关闭掉不需要的文件描述符（即wfd -- pipefd[1]）（当然，父进程也需要关闭不需要的fd -- rfd），在执行完work（子进程的工作）之后，关闭掉rfd（即工作完成了，关闭其管道），然后exit(0)退出进程，等待父进程回收
d. 通过channel控制子进程
// 轮询方案 -- 负载均衡
int NextChannel(int channelnum)
{static int next = 0;int channel = next;next++;next %= channelnum;return channel;
}// 发送相应的任务码到对应管道内
void SendTaskCommand(Channel &channel, int taskCommand)
{write(channel.GetWfd(), &taskCommand, sizeof(taskCommand));
}void CtrlProcessOnce(std::vector<Channel> &channels)
{sleep(1);// a. 选择一个任务int taskcommand = SelectTask();// b. 选择一个信道和进程int channel_index = NextChannel(channels.size());// c. 发送任务SendTaskCommand(channels[channel_index], taskcommand);std::cout << "=================================" << std::endl;std::cout << "taskcommand: " << taskcommand << " channel: " << channels[channel_index].GetName() << " sub process: " << channels[channel_index].GetProcessId() << std::endl;
}void CtrlProcess(std::vector<Channel> &channels, int times = -1)
{if (times > 0){while (times--){CtrlProcessOnce(channels);}}else{while (true){CtrlProcessOnce(channels);}}
}向其发送任务之前，我们需要先选择一个任务，通过随机种子随机数的方式，随机选择我们的一个任务，拿到其任务码（即指针数组下标），然后选择相应的信道和进程（信道和进程一体的），从而向管道发送任务码给子进程
e. 回收管道和子进程
void ClearUpChannel(std::vector<Channel> &channels)
{for (auto &channel : channels){channel.CloseChannel();}for (auto &channel : channels){channel.Wait();}
}我们先将所有的信道关闭，然后在逐个将子进程等待回收
问题1：
那为什么不能边关闭信道边回收呢？？
即
解答1：
在我们创建子进程的过程中，由于父子进程之间的继承，从而导致子进程会拥有父进程的文件描述符内容（即指向同一地方），如果我们边关闭边回收的话，如上图所示，当我们关闭父进程的第一个管道的wfd时，这时候第一个管道的读端的引用计数并未清0，因为子进程2它继承了父进程指向第一个管道的wfd（读端），从而使得读端阻塞，进程不退出，然后wait子进程的时候就会阻塞
在work结束后，才会到下一步close和exit退出子进程；
work结束需要的条件是 n == 0，即读端返回值为0，即
因此上述那种边关闭信道，边wait子进程的方法会阻塞
问题2：
为什么这种方法又能成功回收呢？？
解答2：
因为当我们将所有信道关闭时，关闭到最后一个子进程对应的管道的wfd的时候，该管道的读端的引用计数就会为0，从而读端读到0，该子进程退出，随子进程退出就会使得该子进程指向的前面管道的读端回收，就不会造成前面那种情况
f. 将进程池本身和任务文件本身进行解耦
用回调函数可以很好的改善代码的耦合性
通过文件描述符重定向 dup2，将标准输入（文件描述符 0）重定向到 rfd 所代表的文件，然后再回调task()函数
// 重定向
这样做可以彻底的让我们的子进程执行对应的work时，再也不需要知道有什么管道的读端
(不用管从哪里接收信息，直接认为从标准输入拿到信息即可)
--- 将管道的逻辑和执行任务的逻辑进一步进行解耦
// task_t task : 回调函数
有了它，我们进程池本身的代码和我们任务本身两个文件就彻底解耦了
--- 即既不关心从哪个文件描述符，直接默认从0里面去读，也不关心将来谁调它，因为子进程会自动回调它
3）完整代码
processpool.cc：
#include <iostream>
#include <string>
#include <unistd.h>
#include <vector>
#include <sys/types.h>
#include <sys/wait.h>
#include "Task.hpp"class Channel
{
public:Channel(int wfd, pid_t id, const std::string name): _wfd(wfd), _subprocessid(id), _name(name){}int GetWfd() { return _wfd; }pid_t GetProcessId() { return _subprocessid; }std::string GetName() { return _name; }void CloseChannel() { close(_wfd); }void Wait(){pid_t rid = waitpid(_subprocessid, nullptr, 0);if (rid > 0){std::cout << "wait " << rid << " success" << std::endl;}}~Channel(){}private:int _wfd;int _subprocessid;std::string _name;
};// 形参和命名规范
// const & : 输入型参数
// & : 输入输出型参数
// * : 输出型参数// task_t task : 回调函数
// 有了它，我们进程池本身的代码和我们任务本身两个文件就彻底解耦了
// --- 即既不关心从哪个文件描述符，直接默认从0里面去读，也不关心将来谁调它，因为子进程会自动回调它
void CreateChannelAndSub(int num, std::vector<Channel> *channels, task_t task)
{for (int i = 0; i < num; ++i){// 1.创建管道int pipefd[2] = {0};int n = pipe(pipefd);if (n < 0)exit(1); // 创建管道失败// 2.创建子进程pid_t id = fork();if (id == 0){// Childclose(pipefd[1]);dup2(pipefd[0], 0);task();close(pipefd[0]);exit(0);}// 3.构建一个名字std::string channel_name = "Channel-" + std::to_string(i);// Fatherclose(pipefd[0]);// 拿到了 a.子进程的pid b.父进程需要的管道的w端channels->push_back(Channel(pipefd[1], id, channel_name));}
}int NextChannel(int channelnum) // 轮询方案 -- 负载均衡
{static int next = 0;int channel = next;next++;next %= channelnum;return channel;
}void SendTaskCommand(Channel &channel, int taskCommand)
{write(channel.GetWfd(), &taskCommand, sizeof(taskCommand));
}void CtrlProcessOnce(std::vector<Channel> &channels)
{sleep(1);// a. 选择一个任务int taskcommand = SelectTask();// b. 选择一个信道和进程int channel_index = NextChannel(channels.size());// c. 发送任务SendTaskCommand(channels[channel_index], taskcommand);std::cout << "=================================" << std::endl;std::cout << "taskcommand: " << taskcommand << " channel: " << channels[channel_index].GetName()<< " sub process: " << channels[channel_index].GetProcessId() << std::endl;
}
void CtrlProcess(std::vector<Channel> &channels, int times = -1)
{if (times > 0){while (times--){CtrlProcessOnce(channels);}}else{while (true){CtrlProcessOnce(channels);}}
}void ClearUpChannel(std::vector<Channel> &channels)
{// for (auto &channel : channels)// {// channel.CloseChannel();// channel.Wait();// }// int num = channels.size() -1;// while(num >= 0)// {// channels[num].CloseChannel();// channels[num--].Wait();// }for (auto &channel : channels){channel.CloseChannel();}for (auto &channel : channels){channel.Wait();}
}// ./processpool 3
int main(int argc, char *argv[])
{if (argc != 2){std::cerr << "Usage: " << argv[0] << " processnum" << std::endl;}int num = std::stoi(argv[1]);LoadTask(); // 加载任务std::vector<Channel> channels; // 将管道组织起来// 1.创建信道和子进程CreateChannelAndSub(num, &channels, work);// 2.通过channel控制子进程CtrlProcess(channels, 10);// 3.回收管道和子进程 a.关闭所有的写端 b.回收子进程ClearUpChannel(channels);// // for test// for(auto &channel : channels)// {// std::cout << "====================" << std::endl;// std::cout << channel.GetName() << std::endl;// std::cout << channel.GetWfd() << std::endl;// std::cout << channel.GetProcessId() << std::endl;// }// sleep(100);return 0;
}Task.hpp：
#pragma once#include <iostream>
#include <ctime>
#include <cstdlib>
#include <sys/types.h>
#include <unistd.h>#define TaskNum 3typedef void (*task_t)(); // task_t 函数指针void Print()
{std::cout << "I am print task" << std::endl;
}
void DownLoad()
{std::cout << "I am a download task" << std::endl;
}
void Flush()
{std::cout << "I am a flush task" << std::endl;
}task_t tasks[TaskNum];void LoadTask()
{srand(time(nullptr) ^ getpid() ^ 177); // 种一个随机种子tasks[0] = Print;tasks[1] = DownLoad;tasks[2] = Flush;
}void ExcuteTask(int number)
{if(number < 0 || number > 2) return;tasks[number]();
}int SelectTask()
{return rand() % TaskNum;
}// void work(int rfd)
// {
// while (true)
// {
// int command = 0;
// int n = read(rfd, &command, sizeof(command));
// if (n == sizeof(int))
// {
// std::cout << "pid is: " << getpid() << " handler task" << std::endl;
// ExcuteTask(command);
// }
// else if (n == 0)
// {
// std::cout << "sub process: " << getpid() << " quit" << std::endl;
// break;
// }
// }
// }// 这样做可以彻底的让我们的子进程执行对应的work时，再也不需要知道有什么管道的读端
// (不用管从哪里接收信息，直接认为从标准输入拿到信息即可)
// 将管道的逻辑和执行任务的逻辑进一步进行解耦
void work()
{while (true){int command = 0;int n = read(0, &command, sizeof(command));if (n == sizeof(int)){std::cout << "pid is: " << getpid() << " handler task" << std::endl;ExcuteTask(command);}else if (n == 0){std::cout << "sub process: " << getpid() << " quit" << std::endl;break;}}
}Makefile：
processpool:processpool.ccg++ -o $@ $^ -std=c++11
.PHONY:clean
clean:rm -f processpool