七、进程间通信
1. 进程间通信分类
管道
管道的四种情况：
①管道内部没有数据，并且具有写端的进程没有关闭写端，读端就要阻塞等待，知道管道pipe内部有数据。
②管道内部被写满，并且具有读端的继承没有关闭读端，写端写满管道pipe后，就需要阻塞等待，直到管道清空。
③对于写端而言：关闭了写端管道，读端会将管道pipe中的数据读完，最后会读到返回值0，表示读取完毕。
④对于读端而言：关闭了读端管道，操作系统会直接终止具有写端的进程，通过十三号信号 SIGPIPE 杀掉进程。
管道的五种特性：
①自带同步机制
②通过血缘关系进程进行通信，常见为父子进程
③pipe是面向字节流的
④父子退出，管道自动释放，文件的生命周期是随进程的
⑤管道只能单向通信
我们曾经学的命令行管道 | 本质上就是pipe。
接下来我们根据我们所学的管道知识来实现一个 进程池 。
Makefile：
processpool:processpool.cc g++ -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; // 子进程的id pid_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; }
