【Linux】进程池

目录

进程池

进程池的概念：

手搓进程池：

1、创建信道和子进程

2、通过channel控制子进程

3、回收管道和子进程

---

进程池

进程池的概念：

        定义一个池子，在里面放上固定数量的进程，有需求来了，就拿一个池中的进程来处理任务，等到处理完毕，进程并不关闭，直接将进程再放回进程池中继续等待任务；
        如果有很多任务需要执行，池中的进程数量不够，任务就要等待之前的进程执行任务完毕归来，拿到空闲进程才能继续执行。也就是说，进池中进程的数量是固定的，那么同一时间最多有固定数量的进程在运行；这样不会增加操作系统的调度难度，还节省了开关进程的时间，也一定程度上能够实现并发效果。

看下图，父进程和子进程之间可以通过管道来交互；

如果管道中没有数据，则worker进程就阻塞等待；master向哪个管道写入就唤醒哪一个子进程来处理任务；

手搓进程池：

1、创建信道和子进程

我们用一个类来录父进程读写端的fd和子进程的id,用vector来存储；

• 先来创建一个管道（pipe）
• 管道创建成功后，再创建子进程(fork)
• 关闭不需要的fd
• 将信息存储到vector中

可以将上述代码封装到一个函数中，这样比较好看

#include <iostream>
#include <string>
#include <vector>
#include <unistd.h>
using namespace std;class channel
{
public:channel(int wfd, pid_t id, const string &name) : _wfd(wfd), childid(id), _name(name){}~channel(){}int getwfd() { return _wfd; }pid_t getid() { return childid; }string getname() { return _name; }private:int _wfd;pid_t childid;string _name;
};void work(int rfd)
{
}void create(vector<channel> &channels, int num)
{for (int i = 0; i < num; i++){int pipfd[2] = {0};int n = pipe(pipfd);pid_t id = fork();if (id == 0){// child --readclose(pipfd[1]);work(pipfd[0]);close(pipfd[0]);exit(0);}// father --writeclose(pipfd[0]);string name = "channel-" + to_string(i);channels.push_back(channel(pipfd[1], id, name));close(pipfd[1]);}}int main(int argc, char *argv[])
{if (argc != 2){cerr << "processnum???" << endl;return 1;}int num = stoi(argv[1]);// 1、创建子进程和信道vector<channel> channels;create(channels, num);for (auto channel : channels){cout << channel.getid() << " " << channel.getwfd() << " " << channel.getname() << endl;}}

运行结果：

2、通过channel控制子进程

信道建立好后，下面就是接收主进程给我们的任务就可以了，可是子进程如何接收和识别任务呢？

这里我们可以开一个hpp文件来模拟我们的任务：

.hpp文件是允许声明和实现写到一起的；

 在这个文件中使用函数指针类型来初始化，并且有随机选择任务的函数，执行任务的函数；

#pragma once#include <iostream>
#include <ctime>
#include <stdlib.h>
#include <sys/types.h>
#include <unistd.h>
using namespace std;#define NUM 3
typedef void (*task_t)();task_t tasks[NUM];// 创建任务
void Print()
{cout << "I am Print Task" << endl;
}
void Flush()
{cout << "I am Flush Task" << endl;
}
void Download()
{cout << "I am Download Task" << endl;
}// 初始化
void Loadtask()
{srand(time(nullptr) ^ getpid());tasks[0] = Print;tasks[1] = Download;tasks[2] = Flush;
}void Excutetask(int num)
{if (num < 0 || num > 2)return;tasks[num]();
}int selecttask()
{return rand()%NUM;
}

完成.hpp文件后，在我们的.cpp文件中添加对应的头文件；

• 随机选择一个任务
• 选择信道和进程
• 发送任务----父进程完成write操作，子进程完成read操作

 

运行结果：

3、回收管道和子进程

• 关闭所有w端
• wait，回收子进程

完整代码：

#include <iostream>
#include <string>
#include <vector>
#include <unistd.h>
#include "test.hpp"
#include <sys/types.h>
#include <sys/wait.h>
using namespace std;class channel
{
public:channel(int wfd, pid_t id,const string &name) : _wfd(wfd), childid(id), _name(name){}~channel(){}int getwfd() { return _wfd; }pid_t getid() { return childid; }string getname() { return _name; }void closechannel(){close(_wfd);}void wait(){pid_t rid =waitpid(childid,nullptr,0);if(rid>0){cout<<"wair sucess"<<endl;}}private:int _wfd;pid_t childid;string _name;
};void work(int rfd)
{while (true){int command = 0;int n = read(rfd, &command, sizeof(command));if (n == sizeof(int)){Excutetask(command);}}
}void create(vector<channel> &channels, int num)
{for (int i = 0; i < num; i++){int pipfd[2] = {0};int n = pipe(pipfd);pid_t id = fork();if (id == 0){// child --readclose(pipfd[1]);work(pipfd[0]);close(pipfd[0]);exit(0);}// father --writeclose(pipfd[0]);string name = "channel-";name += to_string(i);channels.push_back(channel(pipfd[1], id, name));}
}int selectchannel(int num)
{static int index = 0;int next = index;index++;index %=num;return next;
}
void send(int selectnum, int channel_index, vector<channel> &channels)
{write(channels[channel_index].getwfd(), &selectnum, sizeof(selectnum));
}void controlonce(vector<channel> &channels)
{// 2.1、选一个任务int selectnum = selecttask();// 2.2、选一个信道和进程int channel_index = selectchannel(channels.size());// 2.3、发送---父进程w，子进程rsend(selectnum, channel_index, channels);cout << "信息发送成功" << endl;
}
void control(vector<channel> &channels, int times = -1)
{if (times > 0){while(times--){controlonce(channels);}}else{while (true){controlonce(channels);}}
}void clean(vector<channel> &channels)
{for(auto channel:channels){channel.closechannel();}for(auto channel:channels){channel.wait();}
}int main(int argc, char *argv[])
{if (argc != 2){cerr << "processnum???" << endl;return 1;}int num = stoi(argv[1]);Loadtask();// 1、创建子进程和信道vector<channel> channels;create(channels, num);// for (auto channel : channels)// {//     cout << channel.getid() << " " << channel.getwfd() << " " << channel.getname() << endl;// }// 2、通过channel控制子进程control(channels, 10);//3、回收管道和子进程clean(channels);
}

以上就是进程池的知识点，希望有所帮助！！！