【Linux】进程间通信(匿/命名管道、共享内存、消息队列、信号量)

文章目录

1. 进程通信的目的
2. 管道
- 2.1 原理
- 2.2 匿名管道
- 2.3 管道通信场景：进程池
- 2.4 命名管道
3. System V共享内存
- 3.1 操作共享内存
- 3.2 使用共享内存通信
4. System V 消息队列（了解）
5. System V 信号量（了解）
- 5.1 信号量
6. IPC的理解
- 6.1 用户角度
- 6.2 内核角度

1. 进程通信的目的

在前面的学习中，我们学到的进程都是孤立的（进程具有独立性），最密切的关系就是父子关系；但是在实际中，并不是单个进程去完成某些任务，进程间也需要协作。

数据传输：一个进程需要将它的数据发送给另一个进程
资源共享：多个进程之间共享同样的资源。
通知事件：一个进程需要向另一个或一组进程发送消息，通知它（它们）发生了某种事件（如进程终止时要通知父进程）。
进程控制：有些进程希望完全控制另一个进程的执行（如Debug进程），此时控制进程希望能够拦截另一个进程的所有陷入和异常，并能够及时知道它的状态改变。

由于进程具有独立性，那么进程间直接通信是不可能的，除非有专门的进程通信技术。

为了解决进程间通信的问题，就引入了进程间通信。

那进程间是如何通信的呢？

本质是：要先让不同的进程看到同一份资源，并且该资源不能是进程的（由于存在写时拷贝），只能是操作系统提供的公共资源。

如果公共资源是文件的形式：管道
如果公共资源是内存块的形式：共享内存
如果公共资源是队列的形式：消息队列
如果是个计数器：信号量

进程间通信有三个标准：

管道
SystemV
POSIX

2. 管道

2.1 原理

首先我们要知道，管道是Unix系统中比较古老的通信方式。

有了文件系统的知识后，我们知道文件加载到内存中的过程入下图：
在这里插入图片描述

如果此时父进程创建子进程，那子进程就需要以父进程的task_struct、file_struct、file等为模板创建自己的；但是文件相关的内容（inode、内核缓冲区）子进程也会创建吗？- - 不会，因为操作系统内不会重复的打开一个文件。

如果内核缓冲区父子进程共用一份，这不就是OS提供的公共资源了吗？
在这里插入图片描述

但是，现在如果让父子进程之间通信，就不需要将内核缓冲区中的内容刷新到磁盘中了；因为不仅逻辑上讲不通，而且还浪费资源。

如果将内核缓冲区设计成一种纯内存级别的，然后用它来专门负责进程间的通信不就可以了吗？- - 是的，我们的管道就是这样产生的。

管道特点：只能进行单向通信（需要关掉相应的文件描述符）
在这里插入图片描述

2.2 匿名管道

内存级别的内核缓冲区就是匿名管道（匿名：无需指定路径、没有名字）。

创建管道的系统调用： pipe

头文件： #include <unistd.h>
功能:创建一无名管道原型 int pipe(int fd[2]);
参数 fd：文件描述符数组,其中fd[0]表示读端, fd[1]表示写端
返回值:成功返回0，失败返回错误代码

在这里插入图片描述

使用pipe实现通信：

#include <iostream>
#include <string>
using namespace std;#include <unistd.h>
#include <sys/types.h>
#include <sys/wait.h>int main()
{int fds[2] = {0};// 1. 创建管道int n = pipe(fds); // fds输出形参数if (n == -1){cerr << "pipe fail" << endl;return 1;}// 2. 创建子进程pid_t id = fork();if (id < 0){return 2;}if (id == 0){// child// 3. 关闭read::close(fds[0]);int count = 0;while (true){string messages = "hello pipe ";messages += to_string(getpid());messages += ",";messages += to_string(count);// 开始通信::write(fds[1], messages.c_str(), messages.size());count++;sleep(1);}exit(0);}else{// father//  3.关闭write::close(fds[1]);char buffer[1024];while(true){size_t n = ::read(fds[0],buffer,1024);if(n > 0){buffer[n] = '\0';cout << "child->father,message:" << buffer << endl;}}pid_t rid = waitpid(id, nullptr, 0);cout << "father wait child success,rid:" << rid << endl;}return 0;
}

管道的四种场景：

管道正常 && 管道为空，read(系统调用)会阻塞
管道正常 && 管道为满，write(系统调用)会阻塞
管道写端关闭，读端继续，读端无内容可读，表示到达文件结尾
管道读端关闭，写端继续，写端无需再写，OS会直接杀掉写端进程（ SIGKILL ）

匿名管道的特性：

面向字节流(不关心对方写的多少)
主要用于具有“血缘关系”的进程进行IPC(Inter-Process Communication)，常用于父子
文件的生命周期随进程，管道也是
管道只能进行单向通信，如若进行双向，那就建立两个管道
管道自带同步互斥等保护机制（对共享资源的）

2.3 管道通信场景：进程池

在我们之前所写的代码中，进程都是各忙各的；那我们能不能将多个进程协同起来，让一个进程向其它进程派发任务呢？
在这里插入图片描述

根据管道的场景：管道正常 && 管道为空，read(系统调用)会阻塞。那么当父进程不向指定的管道中写数据，那与管道相连的进程就必须阻塞。

此时，父进程就可以通过向管道中写数据来控制子进程，此时多进程之间就进行了协同，这种模式就叫做进程池

下面我们就来简单实现一个进程池

先准备一个通用的makefile

BIN = ProcessPool ## binary executable files##---------1. 编译选项---------
CC = g++
#-Wall：warning all 显示所有警告
FLAGS = -c -Wall -std=c++11##---------2. 链接选项---------
LDFLAGS = -o##---------3. 获得相关依赖文件---------
#罗列所有的.cc
SRC = $(shell ls *.cc)
#wildcard是make的函数
#SRC = $(wildcard  *.cc)#形成同名.o
OBJ = $(SRC:.cc=.o)##--------- 4. 执行编译、链接操作---------
$(BIN):$(OBJ)$(CC) $(LDFLAGS) $@ $^
%.o:%.cc$(CC) $(FLAGS) $<.PHONY:clean
clean:rm -f $(BIN) $(OBJ)

接下来，我们要按照下面的步骤书写代码：

先有num个管道，父进程同时对多个管道进行管理
创建num个子进程
关闭相应的读写端
建立信道
父进程向信道中写任务，子进程读任务

Channel.hpp

#include <iostream>
using namespace std;
#include<string>class Channel
{
public:Channel(int wfd,int who):_wfd(wfd),_who(who){_name = "Channel-" + to_string(wfd) + "-" + to_string(who);}~Channel(){}int Get_wfd(){return _wfd;}void Close(){close(_wfd);}string GetName(){return _name;}
private:int _wfd; //标记每个管道的写端string _name; //管道名int _who; 
};

ProcessPoll.cc

#include "Channel.hpp"
#include <vector>#include <unistd.h>
#include <sys/types.h>
enum
{USEERR = 0,PIPEFAIL,FORKFAIL
};void Usage(string ProcessName)
{cout << "Usage:" << ProcessName << " process num" << endl;
}void ExecuteTask()
{}int main(int argc, char *argv[])
{// 检测是否正确使用if (argc != 2){Usage(argv[0]);return USEERR;}int num = stoi(argv[1]); // 获得进程个数vector<Channel> channels(num);// 2. 创建指定个进程for (int i = 0; i < num; i++){// 1.先创建管道int pipefd[2] = {0}; // 返回的两个文件描述符，指向同一个管道int n = ::pipe(pipefd);if (n == -1)return PIPEFAIL;pid_t id = fork();if (id < 0)return FORKFAIL;if (id == 0){// 子进程关闭写端::close(pipefd[1]);// 子进程要关闭 从父进程继承下来的之前的管道if (i > 0){cout << "child " << getpid() << " shut history fd" << endl;for (auto &c : channels){cout << c.Get_wfd() << " ";c.Close();}cout << endl;}// 执行任务dup2(pipefd[0],0);//让所有的子进程都从0里读，无需再传递管道的fdExecuteTask();exit(0);}// 父进程关闭读端::close(pipefd[0]);channels.emplace_back(pipefd[1], id);}return 0;
}

在创建管道和进程时，注意下方情况，需要为子进程关闭不属于它管道的读端，否则在等待子进程时，子进程未到文件尾，不退出，父进程会阻塞式等待。
在这里插入图片描述
此时就会将先前进程的管道的读端在当前进程关闭了

下面使用面向对象的思想将代码拆分一下：

Channel.hpp

#pragma once#include <iostream>
#include <unistd.h>
using namespace std;
#include <string>class Channel
{
public:Channel() = default;Channel(int wfd, pid_t who): _wfd(wfd), _who(who){_name = "Channel-" + to_string(wfd) + "-" + to_string(who);}~Channel(){}int Get_wfd(){return _wfd;}void Close(){close(_wfd);}void Send(int task){write(_wfd,&task,sizeof(task));}pid_t GetPid(){return _who;}string GetName(){return _name;}private:int _wfd;     // 标记每个管道的写端string _name; // 管道名pid_t _who;
};

ProcessPool.hpp

#include "Channel.hpp"
#include "Task.hpp"
#include <vector>
#include <functional>#include <unistd.h>
#include <sys/types.h>
#include <sys/wait.h>// 包装器
using work_t = function<void()>;
enum
{OK = 0,USEERROR,PIPEERROR,FORKERROR
};class ProcessPool
{
public:ProcessPool(int processNum, work_t work): num(processNum), Worker(work){}~ProcessPool(){}int InitProcessPoll(){// 2. 创建指定个进程for (int i = 0; i < num; i++){// 1.先创建管道int pipefd[2] = {0}; // 返回的两个文件描述符，指向同一个管道int n = ::pipe(pipefd);if (n == -1)return PIPEERROR;pid_t id = fork();if (id < 0)return FORKERROR;if (id == 0){// 子进程关闭写端::close(pipefd[1]);// 子进程要关闭 从父进程继承下来的之前的管道if (i > 0){// cout << "child " << getpid() << " shut history fd" << endl;for (auto &c : channels){// cout << c.Get_wfd() << " ";c.Close();}// cout << endl;}// 执行任务dup2(pipefd[0], 0); // 让所有的子进程都从0里读，无需再传递管道的fdWorker();exit(0);}// 父进程关闭读端::close(pipefd[0]);channels.emplace_back(pipefd[1], id);}return OK;}void DispatchTask(){int process = 0;int sum = 20;while (sum--){// a. 选择任务，一个整数int task = taskManeger.Select();// b. 轮循选择一个子进程/管道Channel &cur = channels[process++];process %= channels.size();cout << "-------------------------------" << endl;cout << "send " << task << "to " << cur.GetName() << " 剩余任务：" << sum << endl;cout << "-------------------------------" << endl;// c. 给子进程 派发任务(向相应的管道中写)cur.Send(task);sleep(1);}}void ClearProcessPool(){for (auto &c : channels){c.Close();pid_t rid = waitpid(c.GetPid(), nullptr, 0);if (rid > 0){cout << "father wait pid:" << rid << " success" << endl;}}}// void Debug(vector<Channel> &channels)// {//     for (auto &c : channels)//     {//         cout << "pipe name:" << c.GetName() << endl;//     }// }private:vector<Channel> channels;int num;work_t Worker;
};

Task.hpp

#pragma once#include <unordered_map>
#include <functional>
#include <iostream>
#include <time.h>
#include <stdlib.h>#include <unistd.h>using task_t = std::function<void()>;
static int number = 0;void Print()
{std::cout << "打印任务... pid: " << getpid() << std::endl << std::endl;
}void DownLoad()
{std::cout << "下载任务...pid: " << getpid() << std::endl<< std::endl;
}void SQL()
{std::cout << "数据库任务...pid: " << getpid() << std::endl<< std::endl;
}void Install()
{std::cout << "安装任务...pid: " << getpid() << std::endl<< std::endl;
}class TaskManager
{
public:TaskManager(){srand(time(nullptr));this->Insert(Print);this->Insert(DownLoad);Insert(SQL);Insert(Install);}~TaskManager(){}void Insert(task_t task){tasks[number++] = task;}int Select(){return rand() % number;}void Execute(int num){if (tasks.find(num) == tasks.end())return;tasks[num]();}private:std::unordered_map<int, task_t> tasks;
};TaskManager taskManeger; //创建全局的任务管理对象void Work()
{while (true){int task = 0;int n = read(0, &task, sizeof(task));if (n == 0){cout << "child pid :" << getpid() << " quit success" << endl;break; // 写端关闭，读到了文件尾}else if (n == sizeof(task)){taskManeger.Execute(task);}else{// to dot}}
}

main.cc

#include "ProcessPool.hpp"void Usage(string ProcessName)
{cout << "Usage:" << ProcessName << " process num" << endl;
}int main(int argc, char *argv[])
{// 检测是否正确使用if (argc != 2){Usage(argv[0]);return USEERROR;}int num = stoi(argv[1]); // 获得进程个数ProcessPool* pp = new ProcessPool(num,Work);// 1. 创建进程池pp->InitProcessPoll();//  2. 派发任务pp->DispatchTask();// 3. 清理进程池pp->ClearProcessPool();delete pp;return 0;
}

2.4 命名管道

匿名管道应用的一个限制就是只能在具有共同祖先（具有亲缘关系）的进程间通信。

如果我们想在不相关的进程之间交换数据，可以使用FIFO文件来做这项工作，它经常被称为命名管道。

命名管道是一种特殊类型的文件，创建命名管道的系统命令：mkfifo

在这里插入图片描述

此时，我们就能真正的“看见”一个管道文件了。有了路径和文件名，就能唯一确定一个文件，这样不同的进程就看到了同一份资源了。

它有自己的inode，文件类型为p。
p类型文件：只使用内核文件缓冲区，不做刷新。

在这里插入图片描述

如果我让两个进程一个去命名管道中写，一个读，这不就完成通信了么？

在这里插入图片描述

在一个进程中，也可以使用函数调用：mkfilo，来创建命名管道。

在这里插入图片描述

删除一个命名管道使用：unlink

在这里插入图片描述

由于命名管道有路径和名字，那么不同的进程就可以像打开文件一样使用不同的权限打开命名管道，从而实现进程间通信。

在这里插入图片描述

3. System V共享内存

3.1 操作共享内存

共享内存是最快的IPC形式
一旦这样的内存映射到它的进程的地址空间的共享区，这些进程间数据传递不再涉及到内核。换句话说是进程不再通过执行进入内核的系统调用来传递彼此的数据。

类似于下图：
在这里插入图片描述

那如何在物理内存中得到一块共享内存呢？

创建一个共享内存的系统调用：shmget

在这里插入图片描述

这个函数有三个参数：

key：用来唯一标识该共享内存
size：所创建的共享内存的大小
shmflg：创建共享内存时的选项，常用IPC_CREAT，IPC_EXCL
- IPC_CREAT：如果当前key所对应共享内存不存在，则创建；若存在，返回。保证能返回一个共享内存
- IPC_EXCL：单独使用没有意义。通常IPC_CREAT | IPC_EXCL，如果共享内存不存在，创建它；如果已经存在，则出错返回。保证能返回一个全新的共享内存

返回值：

成功，返回共享内存的id值，该值与key有关
失败，返回-1，错误码被设置

到这里一直有一个疑问，为什么共享内存的key值要用户传递？内核自动生成不香吗？

既然不能让内核生成，那就只能自己创建，并且让这两个进程都能看到。
但是让用户自己设定一个又不好，因为既没有一定的规律，又可能出现大量重复的key，然后导致创建shm失败。
为了解决上述问题，系统提供了一个专门用来生成key的函数：ftok

使用

此时，我们就可以使用shmget创建共享内存了：

在这里插入图片描述

在这里，补充几条关于共享内存的指令：

ipcs -m ：查共享内存
ipcrm -m + shmid：删除shmid对应的共享内存

在这里插入图片描述

既然有了一个共享内存了，那进程应该怎么使用呢？

挂接一个共享内存的系统调用：shmat (at：attach)

在这里插入图片描述
返回值：

成功：返回被挂接的当前进程地址空间的地址
失败：返回（void *） - 1

在这里插入图片描述

为什么挂接失败了呢？

在共享内存的属性中，有一个perms的属性，表示当前shm的权限；没有权限，你怎么挂接呢？
所以，我们需要在创建共享内存时设置权限，可直接在shmflg参数上直接按位与上相应的权限位

在这里插入图片描述

去关联（取消挂接）一个共享内存的系统调用：shmdt (dt：delete attach)

参数：shmat所返回的地址。
在这里插入图片描述

在这里插入图片描述

删除一个共享内存的系统调用：shmctl

在这里插入图片描述
一旦失败，-1会被返回。

在这里插入图片描述

此时，我们共享内存的创建、连接、去连接、删除就搞定了。
在这里插入图片描述

3.2 使用共享内存通信

要想使用共享内存通信，两个进程，进程1先创建shm && 使用；进程2 获取shm && 使用。然后一个进程向所挂接的内存中写，另一个读即可完成通信。

因此，可以将共享内存专门抽离作为一个类。
然后创建全局共享内存的对象，以便进程都能看到

#pragma once
#include <iostream>
#include <string>#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <unistd.h>const std::string gpathName = "/home/wk/linux"; // 路径名可随意指定
const int gpro_id = 101;                        // 项目id随便指定
const int gSize = 4096;
const mode_t gmode = 0600;class ShareMemary
{void CreateHelper(int flag){// 1.获取一个key_key = ftok(gpathName.c_str(), gpro_id);// 2. 创建共享内存_shmid = shmget(_key, gSize, flag);if (_shmid == -1)std::cout << "创建shm失败" << std::endl;elsestd::cout << "创建成功:shmid: " << _shmid << std::endl;}public:ShareMemary(): _shmid(-1), _key(0), _addr(nullptr){}~ShareMemary(){}void CreateSHM(){if (_shmid == -1)CreateHelper(IPC_CREAT | IPC_EXCL | gmode);}void GetSHM(){CreateHelper(IPC_CREAT);}void Attach(){_addr = shmat(_shmid, nullptr, 0);if ((long long)_addr == -1)std::cout << "挂接失败" << std::endl;elseprintf("挂接成功：%p\n", _addr);}void DeAttach(){if (_addr != nullptr)shmdt(_addr);}void DeleteSHM(){int del = shmctl(_shmid, IPC_RMID, nullptr);if (del == -1)std::cout << "删除失败" << std::endl;elsestd::cout << "删除成功" << std::endl;}void* GetAddr(){return _addr;}private:int _shmid;key_t _key;void *_addr;
};ShareMemary shm; //创建shm对象，以便其余进程都能看到

此时，即可完成通信工作。
在这里插入图片描述

但是，此时我们的Server不会像管道那样，没东西可读就阻塞，它反而一直在读；这就有可能造成Client还没写完，Server就读了，也就是没有对资源加保护，有可能造成错误。

我们对加了保护的资源叫做临界资源。
访问公共资源的代码，叫做临界区；未访问公共资源的叫做非临界区。

所以，我们目前可使用管道的特性：管道正常&&管道为空，则阻塞，控制Server读数据的时刻。
因此，可在Server与Client间维护一个管道，Server读，Client写。当Client讲资源完整写道共享内存中，然后向管道发送一个信号，Server从管道中读到该信号就不再read那里阻塞了，此时读共享内存中的资源就是正确的了。

下面的代码模拟了进程间同步的过程。
在这里插入图片描述