---

七、进程间通信

1. 进程间通信分类

管道

管道的四种情况：

①管道内部没有数据，并且具有写端的进程没有关闭写端，读端就要阻塞等待，知道管道pipe内部有数据。
②管道内部被写满，并且具有读端的继承没有关闭读端，写端写满管道pipe后，就需要阻塞等待，直到管道清空。
③对于写端而言：关闭了写端管道，读端会将管道pipe中的数据读完，最后会读到返回值0，表示读取完毕。
④对于读端而言：关闭了读端管道，操作系统会直接终止具有写端的进程，通过十三号信号 SIGPIPE 杀掉进程。

管道的五种特性：

①自带同步机制
②通过血缘关系进程进行通信，常见为父子进程
③pipe是面向字节流的
④父子退出，管道自动释放，文件的生命周期是随进程的
⑤管道只能单向通信

我们曾经学的命令行管道 | 本质上就是pipe。
接下来我们根据我们所学的管道知识来实现一个 进程池 。
Makefile：

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

任务文件 task.hpp：

#pragma once#include <iostream>
#include <unistd.h>
using namespace std;// 函数指针类型
typedef void (*work_t)(int);
typedef void (*task_t)(int, pid_t);void PrintLog(int fd, pid_t pid)
{cout << "sub process: " << pid << ", fd : " << fd << ", task is : print log task\n" << endl;
}void ReloadConf(int fd, pid_t pid)
{cout << "sub process: " << pid << ", fd : " << fd << ", task is : reload conf task\n" << endl;
}void ConnectMysql(int fd, pid_t pid)
{cout << "sub process: " << pid << ", fd : " << fd << ", task is : connect mysql task\n" << endl;
}// 任务列表
task_t tasks[3] = {PrintLog, ReloadConf, ConnectMysql};// 随机选择一个任务
uint32_t NextTask()
{return rand() % 3;
}// 执行任务
void worker(int fd)
{while (true){uint32_t task_id = 0;ssize_t n = read(0, &task_id, sizeof(task_id));if (n == sizeof(task_id)){if (task_id >= 3) continue;tasks[task_id](fd, getpid());}else if (n == 0){cout << "sub process: " << getpid() << " exit" << endl;break;}}
}

进程池主逻辑 processpool.cc：

#include <iostream>
#include <string>
#include <unistd.h>
#include <cstdlib>
#include <vector>
#include <ctime>
#include <sys/wait.h>
#include "task.hpp"
using namespace std;// 枚举错误类型
enum
{UsageError = 1,ArgError,PipeError
};// 打印使用说明
void Usage(const std::string &proc)
{cout << "Usage: " << proc << " number of processes" << endl;
}// 将信道信息封装成一个类
class Channel
{
public:Channel(int wfd, pid_t sub_id, const string &name): _wfd(wfd), _sub_process_id(sub_id), _name(name){}string name(){return _name;}int wfd(){return _wfd;}pid_t pid(){return _sub_process_id;}void Close(){close(_wfd);}~Channel() {}
private:// 信道的写端int _wfd;// 子进程的idpid_t _sub_process_id;// 信道的编号名称string _name;
};// 进程池管理类
class ProcessPool
{
public:ProcessPool(int num_processes): _num_processes(num_processes){}// 创建子进程和信道int CreateProcess(work_t work){for (int i = 0; i < _num_processes; i++){// 创建管道int pipefd[2]{0};int n = pipe(pipefd);if (n < 0)return PipeError;// 创建子进程pid_t id = fork();if (id == 0){// 这里是子进程, 读端close(pipefd[1]);// 这里需要注意的是, 子进程需要从父进程那里接收任务, 所以需要将父进程的写端重定向到标准输入dup2(pipefd[0], 0);// 子进程执行任务work(pipefd[0]);exit(0);}string cname = "Channel-" + to_string(i);// 这里是父进程, 写端close(pipefd[0]);// 放到vector中管理起来_channels.push_back(Channel{pipefd[1], id, cname});}return 0;}// 向下一个信道发送任务(目的是负载均衡)int NextChannel(){static int next = 0;int c = next++;next %= _num_processes;return c;}// 向index进程执行code任务void SendTaskCode(int index, uint32_t code){cout << "send code: " << code << " to " << _channels[index].name() << " sub process id: " << _channels[index].pid() << endl;// 父进程向管道内发送任务，让子进程读取任务write(_channels[index].wfd(), &code, sizeof(code));}// 杀死所有子进程void KillAll(){for (auto& c : _channels){// 父进程关闭写端，子进程读端读到0会自动结束进程c.Close();cout << c.name() << " close done," << " sub process id: " << c.pid() << endl;}}// 等待所有子进程退出void WaitAll(){for (auto& c : _channels){pid_t pid = c.pid();// 回收子进程返回信息pid_t rid = waitpid(pid, nullptr, 0);if (rid == pid){cout << c.name() << " sub process id: " << c.pid() << " exit done" << endl;}}}~ProcessPool() {}
private:// 进程池的大小int _num_processes;// 信道管理容器vector<Channel> _channels;
};// 控制进程池
void CtrlProcessPool(ProcessPool* pp, int cnt)
{while (cnt){// 选择通道int c = pp->NextChannel();// 选择任务uint32_t code = NextTask();// 发送任务到子进程pp->SendTaskCode(c, code);sleep(1);cnt--;}
}int main(int argc, char *argv[])
{if (argc != 2){Usage(argv[0]);return UsageError;}int num_processes = std::stoi(argv[1]);if (num_processes < 1 || num_processes > 5)return ArgError;srand((unsigned)time(nullptr));// 创建进程池对象ProcessPool* pp = new ProcessPool(num_processes);// 创建子进程和信道pp->CreateProcess(worker);// 控制子进程执行指定数量的任务CtrlProcessPool(pp, 10);// 让所有的子进程退出pp->KillAll();// 回收子进程资源pp->WaitAll();return 0;
}

---

未完待续