目录
1.代码介绍
2.channel 类
3.进程池类编写
4.主函数及其他
5. 源码
1.代码介绍
本文代码采用一主多从式（一个主进程（master）多个子进程（worker））通过管道进行通信，实现主进程分发任务，子进程完成任务，当主进程关闭管道时，子进程执行完任务后退出。
2.channel 类
创建一个channel类用于描述子进程，方便主进程与其通信。
成员变量：
int _wfd; //记录管道写描述符用于主进程发送数据
pid_t _pid; //子进程pid
string _name;
代码实现：
class Channel
{
public:
Channel(int wfd, pid_t pid, const string &name) : _wfd(wfd),
_pid(pid), _name(name) {}
~Channel() {}
int wfd() { return _wfd; }
string name() { return _name; }
pid_t pid() { return _pid; }
void Close() { close(_wfd); }//关闭写描述符，子进程读取完毕后会退出
private:
int _wfd;
pid_t _pid;
string _name;
};3.进程池类编写
该类用于管理子进程，具体功能：
1.创建子进程。
2.获取子进程（轮转方式保证负载均衡）。
3.主进程发送任务码给子进程。
4.进程等待。
5.控制进程退出。
代码实现：
class ProcessPool
{
public:
ProcessPool(int num) : _process_num(num) {}
void createProcess(work_t work)
{
vector<int> fds;
for (int i = 0; i < _process_num; ++i)
{
int pipefd[2]{0};
pipe(pipefd);
pid_t pid = fork();
if (pid == 0)
{
if(!fds.empty())
{
for(auto& fd:fds)
{
//关闭之前子进程管道的写端
close(fd);
}
}
close(pipefd[1]);
dup2(pipefd[0], 0);
work();
exit(0);//子进程执行任务完毕会退出
}
close(pipefd[0]);
string cname = "channel-" + to_string(i);
_channels.push_back(Channel(pipefd[1], pid, cname));
fds.push_back(pipefd[1]);
}
}
int NextChannel()
{
static unsigned int index = 0;
return (index++) % _process_num;
}
void SendTaskCode(int index, uint32_t code)
{
cout << "send code: " << code << " to " << _channels[index].name() << " sub prorcess id: " << _channels[index].pid() << endl;
write(_channels[index].wfd(), &code, sizeof(code));
}
void Wait()
{
waitpid(-1, nullptr, 0);
}
void KillAll()
{
for (auto &channel : _channels)
{
channel.Close();
}
}
~ProcessPool() {}
private:
int _process_num;
vector<Channel> _channels;
};4.主函数及其他
主进程发送任务码，通过管道发送给子进程执行对应任务。
void CtrlProcessPool(const shared_ptr<ProcessPool> &processpool_ptr, int cnt)
{
while (cnt)
{
// a. 选择一个进程和通道
int channel = processpool_ptr->NextChannel();
// cout << channel.name() << endl;
// b. 你要选择一个任务
uint32_t code = NextTask();
// c. 发送任务
processpool_ptr->SendTaskCode(channel, code);
sleep(1);
cnt--;
}
}
int main(int argc, char *argv[])
{
if (argc != 2)
{
printf("\n\t usage ./processPool num\n");
return 1;
}
int process_num = stoi(argv[1]);
shared_ptr<ProcessPool> process_ptr = make_shared<ProcessPool>(process_num);
process_ptr->createProcess(worker);
CtrlProcessPool(process_ptr, 5);
process_ptr->KillAll();
process_ptr->Wait();
return 0;
}模拟任务：
#pragma once
#include <iostream>
#include <unistd.h>
#include <functional>
using namespace std;
using work_t = function<void()>;
using task_t = function<void()>;
void PrintLog()
{
cout << "printf log task" << endl;
}
void ReloadConf()
{
cout << "reload conf task" << endl;
}
void ConnectMysql()
{
cout << "connect mysql task" << endl;
}
task_t tasks[3] = {PrintLog, ReloadConf, ConnectMysql};
uint32_t NextTask()
{
return rand() % 3;
}
void worker()
{
// 从0中读取任务即可！
while(true)
{
uint32_t command_code = 0;
ssize_t n = read(0, &command_code, sizeof(command_code));
if(n == sizeof(command_code))
{
if(command_code >= 3) continue;
tasks[command_code]();
}
else if(n == 0) //管道写端关闭读端返回值位0
{
cout << "sub process: " << getpid() << " quit now..." << endl;
break;
}
}
}
5. 源码
processPool.cc
#include <iostream>
#include <string>
#include <cstdlib>
#include <vector>
#include <unistd.h>
#include <ctime>
#include <sys/wait.h>
#include <sys/types.h>
#include <memory>
#include "task.hpp"
using namespace std;
class Channel
{
public:
Channel(int wfd, pid_t pid, const string &name) : _wfd(wfd),
_pid(pid), _name(name) {}
~Channel() {}
int wfd() { return _wfd; }
string name() { return _name; }
pid_t pid() { return _pid; }
void Close() { close(_wfd); }//关闭写描述符，子进程读取完毕后会退出
private:
int _wfd;
pid_t _pid;
string _name;
};
class ProcessPool
{
public:
ProcessPool(int num) : _process_num(num) {}
void createProcess(work_t work)
{
vector<int> fds;
for (int i = 0; i < _process_num; ++i)
{
int pipefd[2]{0};
pipe(pipefd);
pid_t pid = fork();
if (pid == 0)
{
if(!fds.empty())
{
for(auto& fd:fds)
{
close(fd);
}
}
close(pipefd[1]);
dup2(pipefd[0], 0);
work();
exit(0);//子进程执行任务完毕会退出
}
close(pipefd[0]);
string cname = "channel-" + to_string(i);
_channels.push_back(Channel(pipefd[1], pid, cname));
fds.push_back(pipefd[1]);
}
}
int NextChannel()
{
static unsigned int index = 0;
return (index++) % _process_num;
}
void SendTaskCode(int index, uint32_t code)
{
cout << "send code: " << code << " to " << _channels[index].name() << " sub prorcess id: " << _channels[index].pid() << endl;
write(_channels[index].wfd(), &code, sizeof(code));
}
void Wait()
{
waitpid(-1, nullptr, 0);
}
void KillAll()
{
for (auto &channel : _channels)
{
channel.Close();
}
}
~ProcessPool() {}
private:
int _process_num;
vector<Channel> _channels;
};
void CtrlProcessPool(const shared_ptr<ProcessPool> &processpool_ptr, int cnt)
{
while (cnt)
{
// a. 选择一个进程和通道
int channel = processpool_ptr->NextChannel();
// cout << channel.name() << endl;
// b. 你要选择一个任务
uint32_t code = NextTask();
// c. 发送任务
processpool_ptr->SendTaskCode(channel, code);
sleep(1);
cnt--;
}
}
int main(int argc, char *argv[])
{
if (argc != 2)
{
printf("\n\t usage ./processPool num\n");
return 1;
}
int process_num = stoi(argv[1]);
shared_ptr<ProcessPool> process_ptr = make_shared<ProcessPool>(process_num);
process_ptr->createProcess(worker);
CtrlProcessPool(process_ptr, 5);
process_ptr->KillAll();
process_ptr->Wait();
return 0;
}
task.hpp:
#pragma once
#include <iostream>
#include <unistd.h>
#include <functional>
using namespace std;
using work_t = function<void()>;
using task_t = function<void()>;
void PrintLog()
{
cout << "printf log task" << endl;
}
void ReloadConf()
{
cout << "reload conf task" << endl;
}
void ConnectMysql()
{
cout << "connect mysql task" << endl;
}
task_t tasks[3] = {PrintLog, ReloadConf, ConnectMysql};
uint32_t NextTask()
{
return rand() % 3;
}
void worker()
{
// 从0中读取任务即可！
while(true)
{
uint32_t command_code = 0;
ssize_t n = read(0, &command_code, sizeof(command_code));
if(n == sizeof(command_code))
{
if(command_code >= 3) continue;
tasks[command_code]();
}
else if(n == 0) //管道写端关闭读端返回值位0
{
cout << "sub process: " << getpid() << " quit now..." << endl;
break;
}
}
}
