当前位置： 首页 > news >正文

深入探索Linux：忙碌的车间“进程”间通信

news 2025/7/29 5:45:27

目录

理解层面

为什么要进程间通信？

怎么通信？

什么是通信？

环境问题

具体通信方式的原理+代码

管道

什么是管道

匿名管道

管道管理

demo代码

管道5种特性：

4种通信情况：

管道的容量

管道的写入原子性

基于匿名管道应用场景---进程池

ProcessPool.hpp:

命名管道

server.cc:

client.cc:

理解层面

为什么要进程间通信？

  • 数据传输：⼀个进程需要将它的数据发送给另⼀个进程

  • 资源共享：多个进程之间共享同样的资源。

  • 通知事件：⼀个进程需要向另⼀个或⼀组进程发送消息，通知它（它们）发生了某种事件（如进程终止时要通知父进程）。

  • 进程控制：有些进程希望完全控制另⼀个进程的执行（如Debug进程），此时控制进程希望能够拦截另⼀个进程的所有陷入和异常，并能够及时知道它的状态改变。

怎么通信？

进程间通信的本质：是先让不同的进程，先看到同一份资源[“内存”]（然后才有通信的条件）

  • 这个资源不会是任何一个进程提供的。比如：你拥有父子进程，在父进程里new、malloc一段空间想交给子进程，这是做不到的，因为父子进程之间会发生写时拷贝。它们两个管理的不可能会是同一块空间。

  • 块资源是由OS提供的。既然是要OS提供的，那么势必要求OS给我们提供创建这个资源，使用这个资源，销毁这个资源这样的操作。要使用这样的操作，就势必要提供对应的系统调用接口，只要是系统调用接口，就势必要被操作系统的编写者进行规范统一的设计

什么是通信？

进程间通信指的是运行在一个计算机或不同计算机上的多个进程之间进行数据交换和通信的技术

环境问题

ubuntu 20.04 + c++ + vscode

具体通信方式的原理+代码

管道

什么是管道

  • 管道是Unix中最古老的进程间通信的形式
  • 我们把从⼀个进程连接到另⼀个进程的⼀个数据流称为⼀个“管道”

匿名管道

匿名管道，通常用来做父子通信

在父进程创建子进程时，子进程是否需要将父进程打开的文件也拷贝一份？

  • 不用，当父进程创建一个子进程时，代码和数据是以写时拷贝的方式拷贝的，而对于父进程打开的文件，是以浅拷贝的方式将文件指针进行拷贝，比如父进程打开的三个标准流，子进程会拷贝*file，也就是说，父子进程指向的标准流是同一个

管道管理

管道也是文件，属于文件系统

真的管道不需要刷新到磁盘，和磁盘没关系

我们是怎么保存两个进程打开的是同一个管道文件？

  • 子进程继承文件描述符表

demo代码

​
//最开始的时候，不想为单独为进程通信另起炉灶，就想基于文件系统来搞，所以这就是基于文件系统与操作系统的一些特性搞出来的东西
#include<iostream>
#include<unistd.h>
#include<cstdio>
#include<cstring>
#include<sys/types.h>
#include<sys/wait.h>void ChildWrite(int wfd)
{char buffer[1024];int cnt=0;while(true){snprintf(buffer,sizeof(buffer),"I am chile, pid: %d, cnt: %d",getpid(),cnt++);write(wfd,buffer,strlen(buffer));sleep(1);}
}void FatherRead(int rfd)
{char buffer[1024];while(true){buffer[0]=0;ssize_t n=read(rfd,buffer,sizeof(buffer)-1);if(n>0){buffer[n]=0;std::cout<<"child say:"<<buffer<<std::endl;}}
}int main()
{//创建管道int fds[2]={0}; //fds[0]:读端 fds[1]:写端 int n=pipe(fds);if(n<0){std::cerr<<"pipe error"<<std::endl;return 1;}std::cout<<"fds[0]: "<<fds[0]<<std::endl;std::cout<<"fds[1]: "<<fds[1]<<std::endl;//创建子进程pid_t id=fork();if(id==0){//关闭不需要的读写端，形成单项通信的信道close(fds[0]);ChildWrite(fds[1]);close(fds[1]);exit(0);}//关闭不需要的读写端，形成单项通信的信道close(fds[1]);FatherRead(fds[0]);waitpid(id,nullptr,0);close(fds[0]);return 0;
}​

这种通信方式就叫做管道

基于匿名管道应用场景---进程池

管道5种特性：

  • 匿名管道，只能用来进行具有血缘关系的进程进行进程间通信（常用于父子）
  • 管道文件，自带同步机制。子进程每隔一秒写入数据，父进程立即读取数据，当父进程将文件中子进程输入的数据读取到最后完后，会进入阻塞状态，等待子进程输入数据。这一特性被叫做同步特性，也叫同步机制
  • 管道是面向字节流的
  • 管道是单项通信的，属于半双工的一种特殊情况

  1. 任何时刻，一个发，一个收----半双工

  2. 任何时刻，同时收发----全双工

  • 管道）文件的生命周期，是随进程的。当父子进程都结束后，OS会根据管道文件的引用计数，来判断该管道文件是否要关闭。

4种通信情况：

  • 写慢，读快：读端就要阻塞（进程阻塞），其实就是在等待写端写入数据

  • 写快，读慢：写满了的时候，写端就会阻塞，其实就是在等待读端读取数据

  • 写端关闭，读端继续读：read读到返回值为0，表示读到文件结尾

  • 读关闭，写端继续写：这种情况下，写端写入没有任何意义且浪费空间，同时OS不会做没有意义且浪费空间的事，所以OS会关闭写端进程，并发送异常信号

管道的容量

  • 在ubuntu下，管道大小为64kb

管道的写入原子性

  • 简单讲就是只管结果，不管过程

基于匿名管道应用场景---进程池

和vector扩容类似，每次扩容都会多扩一些，也就相当于提前开一部分空间。

这里就是提前创建一批子进程。

ProcessPool.hpp:
#ifndef __PROCESS_POOL_HPP__
#define __PROCESS_POOL_HPP__#include <iostream>
#include <vector>
#include <unistd.h>
#include "Task.hpp"
#include "sys/wait.h"// 先将管道描述起来
class Channel
{
public:Channel(int fd, pid_t id) : _wfd(fd), _subid(id){_name = "channel-" + std::to_string(_wfd) + "-" + std::to_string(_subid);}void Send(int code){int n = write(_wfd, &code, sizeof(code));(void)n;}int wfd() { return _wfd; }pid_t SubId() { return _subid; }std::string name() { return _name; }void Close(){close(_wfd);}void Wait(){pid_t rid = waitpid(_subid, nullptr, 0);(void)rid;}~Channel() {}private:int _wfd;pid_t _subid;std::string _name;
};// 再将管道组织起来
class ChannelManager
{
public:ChannelManager() : _next(0){}void Insert(int wfd, pid_t subid){// 构建channel并进行管理_channels.emplace_back(wfd, subid);// Channel c(wfd,subid);// _channel.push_back(c);}Channel &Select(){auto &c = _channels[_next];_next++;_next %= _channels.size();return c;}void PrintChannel(){for (auto &channel : _channels){std::cout << channel.name() << std::endl;}}void CloseAll(){for (auto &channel : _channels){channel.Close();} }void StopSubProcess(){for (auto &channel : _channels){channel.Close();std::cout << "关闭:" << channel.name() << std::endl;}}void WaitSubProcess(){for (auto &channel : _channels){channel.Wait();std::cout << "回收:" << channel.name() << std::endl;}}void CloseAndWait(){// for (auto &channel : _channels)// {//     channel.Close();//     std::cout << "关闭:" << channel.name() << std::endl;//     channel.Wait();//     std::cout << "回收:" << channel.name() << std::endl;// }/*当对子进程进行回收时，如果将关闭子进程与会收子进程放到同一个语法糖下进行循环遍历Stop，就会造成子进程无限阻塞的情况*/// 解决方案1：// for(int i=_channels.size()-1;i>=0;i--)// {//     _channels[i].Close();//     std::cout << "关闭:" << _channels[i].name() << std::endl;//     _channels[i].Wait();//     std::cout << "回收:" << _channels[i].name() << std::endl;// }//解决方案2：for (auto &channel : _channels){channel.Close();std::cout << "关闭:" << channel.name() << std::endl;channel.Wait();std::cout << "回收:" << channel.name() << std::endl;}}~ChannelManager() {}private:// 将管道组织成数组std::vector<Channel> _channels;int _next;
};const int gdefaultnum = 5;class ProcseePool
{
public:ProcseePool(int num) : _process_num(num){_tm.Register(PrintLog);_tm.Register(Download);_tm.Register(Upload);}void work(int rfd){while (true){int code = 0;ssize_t n = read(rfd, &code, sizeof(code));if (n > 0){if (n != sizeof(code)){continue;}std::cout << "子进程[" << getpid() << "]收到一个任务码" << code << std::endl;_tm.Execute(code);}else if (n == 0){std::cout << "子进程退出" << std::endl;break;}else{std::cout << "读取错误" << std::endl;break;}}}bool Create(){// 创建进程池for (int i = 0; i < _process_num; i++){// 创建管道int pipefd[2] = {0};int n = pipe(pipefd);if (n < 0){return false;}// 创建子进程pid_t subid = fork();if (subid < 0){// 创建失败return false;}else if (subid == 0){   /*创建子进程后，让子进程从父进程那里继承下来的哥哥进程的写端文件描述符关闭*/_cm.CloseAll();// 子进程// 子进程读，关闭不必要的文件描述符close(pipefd[1]);work(pipefd[0]);close(pipefd[0]);exit(0);}else{// 父进程// 父进程写，关闭不必要的文件描述符close(pipefd[0]);_cm.Insert(pipefd[1], subid);}}return true;}void Debug(){_cm.PrintChannel();}void Run(){// 选择一个任务int taskcode = _tm.Code();// 选择一个信道(子进程)，负载均衡的选择一个子进程，即不出现让一个子进程忙死，其他子进程闲死的情况// 1.轮询// 2.随机// 3.channel添加负载指标// 这里使用轮询auto &c = _cm.Select();std::cout << "选择了一个子进程:" << c.name() << std::endl;// 发送任务c.Send(taskcode);std::cout << "发送了一个任务码:" << taskcode << std::endl;}void Stop(){// 关闭父进程的所有wfd// _cm.StopSubProcess();// _cm.WaitSubProcess();_cm.CloseAndWait();}~ProcseePool(){}private:// 进程池中包含了管道ChannelManager _cm;int _process_num; // 创建进程的数量TaskManager _tm;
};#endif

​

#pragma once#include <iostream>
#include <vector>
#include <ctime>
typedef void (*task_t)();/////////////////////debug////////////////////
void PrintLog()
{std::cout<<"我是一个打印日志的任务"<<std::endl;
}void Download()
{std::cout<<"我是一个下载的任务"<<std::endl;
}void Upload()
{std::cout<<"我是一个上传的任务"<<std::endl;
}
////////////////////////////////////////////////class TaskManager
{
public:TaskManager(){srand(time(nullptr));}//注册任务void Register(task_t t){_tasks.push_back(t);}//任务码int Code(){return rand() % _tasks.size();}//执行任务void Execute(int code){if(code >= 0 && code <_tasks.size()){_tasks[code]();}}~TaskManager(){}private:std::vector<task_t> _tasks;
};
#include"ProcessPool.hpp"
int main()
{ProcseePool pp(gdefaultnum);pp.Create();int cnt=10;// pp.Debug();while(cnt--){pp.Run();sleep(1);}pp.Stop();return 0;
}

这个在上面代码中都有写

命名管道

我们知道了，具有血缘关系的进程之间可以通过匿名管道进行通信，那不具有血缘关系的进程之间该如何通信？

为什么给出路径后，能肯定打开是同一份文件？

  • 因为路径具有唯一性。

  • 文件具有文件名，在同一个目录下不会出现重名的文件

命名管道就是这个原理，只不过命名管道文件不是普通文件

补充图中的内容：

OS发现打开的是同一份文件后，不会把文件加载两次，这就是为什么进程B的struct file与进程A的struct file会指向同一个inode与文件缓冲区。

命名管道文件：fifo

删除命名管道：可以用rm 指令或者unlink

代码实现进程间命名管道通信：

server.cc:
#include <iostream>
#include <string>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <unistd.h>
#include "comm.hpp"int main()
{// 创建管道文件umask(0);int n = mkfifo(FIFO_FILE, 0666);if (n != 0){std::cerr << "mkfifo" << std::endl;return 1;}int fd = open(FIFO_FILE, O_RDONLY);if (fd < 0){std::cerr << "open" << std::endl;return 2;}char buffer[1024];while (true){int n = read(fd, buffer, sizeof(buffer) - 1);if (n > 0){buffer[n] = 0;std::cout << "client say:" << buffer << std::endl;}}close(fd);return 0;
}
client.cc:
#include <iostream>
#include <string>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <unistd.h>
#include "comm.hpp"int main()
{int fd = open(FIFO_FILE, O_WRONLY);if (fd < 0){std::cerr << "open" << std::endl;return 2;}while (true){std::cout<<"Please Enter#";std::string message;std::cin>>message;int n = write(fd, message.c_str(), message.size());if (n > 0){}}close(fd);return 0;
}

comm.hpp:

#pragma once#define FIFO_FILE "fifo"

http://www.xdnf.cn/news/1202293.html

相关文章：

  • 四、计算机组成原理——第6章：总线
  • 微信小程序——早餐小程序
  • LeetCode 85. 最大矩形
  • 「源力觉醒 创作者计划」_文心大模型4.5系列开源模型，意味着什么？对开发者、对行业生态有何影响？
  • SpringBoot 发送邮件
  • Datawhale AI夏令营--Task2：理解项目目标、从业务理解到技术实现！
  • 数值计算 | 图解基于龙格库塔法的微分方程计算与连续系统离散化(附Python实现)
  • MQTT之“SUBSCRIBE报文和SUBACK报文”
  • “太赫兹”
  • 【华为机试】210. 课程表 II
  • 自动化测试常用函数
  • XML Expat Parser：深入解析与高效应用
  • 【CDA干货】金融超市电商App经营数据分析案例
  • 写一个3D旋转的python程序
  • 字节跳动开源Coze，开启AI Agent开发新时代？
  • 【Linux篇章】穿越数据迷雾：HTTPS构筑网络安全的量子级护盾，重塑数字信任帝国！
  • 新能源行业B端极简设计：碳中和目标下的交互轻量化实践
  • 【数据架构09】人工智能及数据智能架构篇
  • 群晖Synology Drive：打造高效安全的私有云协作平台
  • 优测推出HarmonyOS全场景测试服务，解锁分布式场景应用卓越品质!
  • httpx 接口测试教程
  • HarmonyOS 6 云开发-用户头像上传云存储
  • 打通视频到AI的第一公里：轻量RTSP服务如何重塑边缘感知入口？
  • UniappDay04
  • Java 排序
  • Kafka——请求是怎么被处理的？
  • flutter使用firebase集成谷歌，苹果登录
  • Claude Launcher：支持Kimi K2的Claude Code可视化启动工具
  • 力扣988. 从叶结点开始的最小字符串
  • 负载均衡集群HAproxy