目录
进程池
进程池的概念：
手搓进程池：
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 --read
close(pipfd[1]);
work(pipfd[0]);
close(pipfd[0]);
exit(0);
}
// father --write
close(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 --read
close(pipfd[1]);
work(pipfd[0]);
close(pipfd[0]);
exit(0);
}
// father --write
close(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，子进程r
send(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);
}以上就是进程池的知识点，希望有所帮助！！！
