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

进程间通信--匿名管道

article 2025/3/16 6:58:28

进程间通信介绍

 进程间通信目的

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

 管道


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

 用fork来共享管道原理

 站在文件描述符角度-深度理解管道 （内存级）

 左边是进程管理，右边是文件管理，进程通过全局变量能找到文件描述符标，也就找到了对应的file文件也能打开磁盘上的文件拷贝到缓冲区里进行读写，父进程创建子进程，发生写实拷贝，类似于浅拷贝，此时不做文件操作，文件描述符里面的对于关系和父进程一样，struc_file没有关闭，因为父进程还在， struc_file的对应的引用计数不为0就一定不会关闭，既然访问的同一个struc_file，也可以同时访问其中的file文件，通过缓冲区可以进行文件的读写

  1. 父进程通过读写两种方式打开内存级的文件返回给上层完成管道的创建
  2. 子进程继承父进程的文件描述符表，发生浅拷贝，也能拿到父进程以读写打开的管道文件
  3. 父子都看得到，让父子单向通信，父进程写，子进程读，各自关闭掉自己不需要的文件描述符

这个是管道是OS单独设计的，得配上单独的系统调用：pipe 内存级的，不需要文件路径，没有文件名，所以叫匿名管道，那我们怎么保证，两个进程打开的是同一个管道的？

子进程继承了父进程的文件描述符表

站在内核角度-管道本质

 匿名管道

#include <unistd.h>
功能:创建⼀⽆名管道
原型
int pipe(int fd[2]);
参数
fd：⽂件描述符数组,其中fd[0]表⽰读端, fd[1]表⽰写端
返回值:成功返回0，失败返回错误代码
#include<iostream>
#include<unistd.h>
using namespace std;
int main()
{
    int fds[2]={0};

    int n=pipe(fds);
    if(n<0)
    {
        cerr<<"Pipe error"<<endl;
        return 1;
    }
    cout<<"fds[0]"<<fds[0]<<endl;
    cout<<"fds[1]"<<fds[1]<<endl;
   
    return 0;
}

0，1，2被三个标准占用了，从3，4开始

父写子读，实现通信

#include<iostream>
#include<unistd.h>
#include<sys/wait.h>
#include<sys/types.h>
#include<cstring>
using namespace std;

void ChildWrite(int fd)
{
char buffer[1024];
int cnt=0;
while(true)
{
snprintf(buffer,sizeof(buffer),"I am child,pid:%d cnt:%d\n",getpid(),cnt++); 
write(fd,buffer,strlen(buffer));
sleep(1);
}
}
void FatherRead(int fd)
{
    char buffer[1024];
 while(true)
 {
    buffer[0]=0;
    ssize_t n=read(fd,buffer,sizeof(buffer)-1);
    if(n>0)
    {
        buffer[n]=0;
        cout<<"child says: "<<buffer<<endl;
    }
 }
}
int main()
{  //1.创建管道
    int fds[2]={0}; //fds[0]读端，fds[1]写端

    int n=pipe(fds);
    if(n<0)
    {
        cerr<<"Pipe error"<<endl;
        return 1;
    }
    cout<<"fds[0]"<<fds[0]<<endl;
    cout<<"fds[1]"<<fds[1]<<endl;
   //3.创建子进程 f->r c->w
    pid_t id=fork();
  if(id==0)
  {
    //child
    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;
}

父进程写的cnt在不断变化，子进程能读到，说明实现了管道通信

五种特性

1.匿名管道，只能用来进行具有血缘关系的进程进行进程间通信(常用父与子，如上述代码)

2.管道文件，自带同步机制（父子进程进行IO同时进行，一个读一个写，不管是父快还是子快，父不断写，子read读不到会阻塞住直到读到为止）

3.管道是面向字节流的

4.管道是单向通信的（要么父写子读，要么子写父读）

5. （管道）文件的生命周期随进程

4种通信情况 

1.写慢，读快------>读端阻塞等待写端(进程)

2.写快，读慢------>缓冲区写满了，写要阻塞等待读端

3.写关，读开------>read会读到返回值0，表示文件结尾

//写一条就关
void ChildWrite(int fd)
{
char buffer[1024];
int cnt=0;
while(true)
{
snprintf(buffer,sizeof(buffer),"I am child,pid:%d cnt:%d\n",getpid(),cnt++); 
write(fd,buffer,strlen(buffer));
sleep(1);
break;
}
}
//观察n的返回值
void FatherRead(int fd)
{
    char buffer[1024];
 while(true)
 {
    buffer[0]=0;
    ssize_t n=read(fd,buffer,sizeof(buffer)-1);
    if(n>0)
    {
        buffer[n]=0;
        cout<<"child says: "<<buffer<<endl;
    }
    else
    {
      cout<< "n:"<<n<<endl;
    }
 }
}

 4.读关，写开------->写端再写没有意义，OS会杀掉写端进程

#include<iostream>
#include<unistd.h>
#include<sys/wait.h>
#include<sys/types.h>
#include<cstring>
using namespace std;

void ChildWrite(int fd)
{
char buffer[1024];
int cnt=0;
while(true)
{
snprintf(buffer,sizeof(buffer),"I am child,pid:%d cnt:%d\n",getpid(),cnt++); 
write(fd,buffer,strlen(buffer));
sleep(1);
}
}
void FatherRead(int fd)
{
    char buffer[1024];
 while(true)
 {
    buffer[0]=0;
    ssize_t n=read(fd,buffer,sizeof(buffer)-1);
    if(n>0)
    {
        buffer[n]=0;
        cout<<"child says: "<<buffer<<endl;
    }
    else
    {
      cout<< "n:"<<n<<endl;
      sleep(4);
    }
    break;
 }
}
int main()
{  //1.创建管道
    int fds[2]={0}; //fds[0]读端，fds[1]写端

    int n=pipe(fds);
    if(n<0)
    {
        cerr<<"Pipe error"<<endl;
        return 1;
    } 
    cout<<"fds[0]"<<fds[0]<<endl;
    cout<<"fds[1]"<<fds[1]<<endl;
   //3.创建子进程 f->r c->w
    pid_t id=fork();
  if(id==0)
  {
    //child
    close(fds[0]); //关闭读端
    ChildWrite(fds[1]);
    close(fds[1]);
    exit(0);
  }
close(fds[1]); //关闭写端
FatherRead(fds[0]);
close(fds[0]);
  int status;
int ret=waitpid(id,&status,0);
//获取到子进程的退出状态
if(ret>0)
{
  printf("child code: %d exited  status: %d\n",(status)>>8&0xff,(status)&0x7f);
}
return 0;
}

发送异常信号13 SIGPIPE 

 基于匿名管道----进程池

父进程创建多个子进程，用匿名管道分派任务

 1.构建进程链，为进程池做准备

#ifndef __PROCESS__POOL_HPP__
#define __PROCESS__POOL_HPP__
#include <iostream>
#include <vector>
#include <unistd.h>
#include<cstdlib>
using namespace std;
const int gdefaultnum = 5; // 要创建几个进程
// 先描述 单个进程
class Channel
{
public:
    Channel() {}
    ~Channel() {}

private:
    int _wfd;
};

// 在组织  进程链
class ChannelManager
{
public:
    ChannelManager() {}
    ~ChannelManager() {}

private:
    vector<Channel> _channels;
};

 2.创建进程池，提供管道条件

// 进程池
class ProcessPool
{
public:
    ProcessPool(int num) : _process_num(num) {}
    ~ProcessPool() {}
    bool Create()
    {
        int pipefd[2] = {0};
        for (int i = 0; i < _process_num; i++)
        {
            // 1.创建管道

            int n = pipe(pipefd);
            if (n < 0)
                return false;
        }
        // 2.创建子进程  各自关闭不需要的文件描述符
        pid_t id = fork();
        if (id < 0)
            return false;
        else if (id == 0)
        {
            // 子进程 --->读
            // 3.关闭不需要的文件描述符
            close(pipefd[1]);
            exit(0);
        }
        else
        {
            // 父进程 --->写
            // 3.关闭不需要的文件描述符
            close(pipefd[0]);
        }
        return true;
    }

private:
    ChannelManager _cm; // 进程链
    int _process_num;   // 进程个数
};

3.父子各自打印验证是否通信

void BuildChannel(int wfd, pid_t subid)
    {
        _channels.emplace_back(wfd, subid);
        // Channel c(wfd,subid);
        // _channels.push_back(c);
    }
  
void Print()
{
    for(auto  &chnnel : _channels)
    {
cout<<chnnel.Name()<<endl;
    }
}

void Work(int rfd)
    {
        while (true)
        {
            cout << "我是子进程，我的rfd是：" << rfd << endl;
            sleep(2);
        }
    }
//ProcessPool.hpp
#ifndef __PROCESS__POOL_HPP__
#define __PROCESS__POOL_HPP__
#include <iostream>
#include <vector>
#include <unistd.h>
#include <cstdlib>
using namespace std;
const int gdefaultnum = 5; // 要创建几个进程
// 先描述 单个进程
class Channel
{
public:
    Channel(int fd, pid_t id) : _wfd(fd), _subid(id) { _name = "chnnel-" + std::to_string(_wfd) + "-" + std::to_string(_subid); }
    ~Channel() {}
    int Fd(){return  _wfd;}
    pid_t Subid(){return  _subid;}
    string Name(){return  _name;}
private:
    int _wfd;
    pid_t _subid;
    std::string _name;
};

// 在组织  进程链
class ChannelManager
{
public:
    ChannelManager() {}
    ~ChannelManager() {}
    void BuildChannel(int wfd, pid_t subid)
    {
        _channels.emplace_back(wfd, subid);
        // Channel c(wfd,subid);
        // _channels.push_back(c);
    }
  
void Print()
{
    for(auto  &chnnel : _channels)
    {
cout<<chnnel.Name()<<endl;
    }
}
private:
    vector<Channel> _channels;
};

// 进程池
class ProcessPool
{
public:
    ProcessPool(int num) : _process_num(num) {}
    ~ProcessPool() {}

    void Work(int rfd)
    {
        while (true)
        {
            cout << "我是子进程，我的rfd是：" << rfd << endl;
            sleep(2);
        }
    }
    bool Create()
    {

        for (int i = 0; i < _process_num; i++)
        {
            int pipefd[2] = {0};
            // 1.创建管道

            int n = pipe(pipefd);
            if (n < 0)
                return false;

            // 2.创建子进程  父子各自关闭不需要的文件描述符
            pid_t id = fork();
            if (id < 0)
                return false;
            else if (id == 0)
            {
                // 子进程 --->读
                // 3.关闭不需要的文件描述符
                close(pipefd[1]);
                Work(pipefd[0]);
                exit(0);
            }
            else
            {
                // 父进程 --->写
                // 3.关闭不需要的文件描述符
                close(pipefd[0]);
                _cm.BuildChannel(pipefd[1], id);
                close(pipefd[1]);
            }
            
        }
        return true;
    }
  void Debug()
  {
    _cm.Print();
  }
private:
    ChannelManager _cm; // 进程链
    int _process_num;   // 进程个数
};

#endif
//Main.cc

#include"ProcessPool.hpp"

int main()
{
    ProcessPool pp(gdefaultnum);
    //创建进程池
    pp.Create();
    //打印进程池
    pp.Debug();
    sleep(1000);
    return 0;
}

实现通信 

 4.分配任务，子写父读

 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;
                }
                cout << "子进程[]"<<getpid()<<"]收到一个任务码:" << code << endl;
            }
            else if (n == 0)
            {
                cout << "子进程退出" << endl;
                break;
            }
            else
            {
                // 读失败
                cout << "读取错误" << endl;
                break;
            }
        }
    }
 void PushTack(int taskcode)
    {
        // 1.选择一个子进程，采用轮询，防止负载均衡和负载不均衡
        auto &c = _cm.Select();
        cout << "选择一个子进程：" << c.Name() << endl;
        // 2.发送任务
        c.Send(taskcode);
        cout << "发送了一个任务码：" << taskcode << endl;
    }

完整代码

//ProcessPool.hpp
#ifndef __PROCESS__POOL_HPP__
#define __PROCESS__POOL_HPP__
#include <iostream>
#include <vector>
#include <unistd.h>
#include <cstdlib>
using namespace std;
const int gdefaultnum = 5; // 要创建几个进程
// 先描述 单个进程
class Channel
{
public:
    Channel(int fd, pid_t id) : _wfd(fd), _subid(id) { _name = "chnnel-" + std::to_string(_wfd) + "-" + std::to_string(_subid); }
    ~Channel() {}
    int Fd() { return _wfd; }
    pid_t Subid() { return _subid; }
    string Name() { return _name; }
    void Send(int code)
    {
        int n = write(_wfd, &code, sizeof(code));
        (void)n;
    }

private:
    int _wfd;
    pid_t _subid;
    std::string _name;
};

// 在组织  进程链
class ChannelManager
{
public:
    ChannelManager() : _next(0) {}
    ~ChannelManager() {}
    void BuildChannel(int wfd, pid_t subid)
    {
        _channels.emplace_back(wfd, subid);
        // Channel c(wfd,subid);
        // _channels.push_back(c);
    }
    // 轮询
    Channel &Select()
    {
        auto &c = _channels[_next];
        _next++;
        _next %= _channels.size();
        return c;
    }
    void Print()
    {
        for (auto &chnnel : _channels)
        {
            cout << chnnel.Name() << endl;
        }
    }

private:
    vector<Channel> _channels;
    int _next;
};

// 进程池
class ProcessPool
{
public:
    ProcessPool(int num) : _process_num(num) {}
    ~ProcessPool() {}

    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;
                }
                cout << "子进程[]"<<getpid()<<"]收到一个任务码:" << code << endl;
            }
            else if (n == 0)
            {
                cout << "子进程退出" << endl;
                break;
            }
            else
            {
                // 读失败
                cout << "读取错误" << endl;
                break;
            }
        }
    }
    bool Create()
    {

        for (int i = 0; i < _process_num; i++)
        {
            int pipefd[2] = {0};
            // 1.创建管道

            int n = pipe(pipefd);
            if (n < 0)
                return false;

            // 2.创建子进程  父子各自关闭不需要的文件描述符
            pid_t id = fork();
            if (id < 0)
                return false;
            else if (id == 0)
            {
                // 子进程 --->读
                // 3.关闭不需要的文件描述符
                close(pipefd[1]);
                Work(pipefd[0]);
                exit(0);
            }
            else
            {
                // 父进程 --->写
                // 3.关闭不需要的文件描述符
                close(pipefd[0]);
                _cm.BuildChannel(pipefd[1], id);
                close(pipefd[1]);
            }
        }
        return true;
    }

    void Debug()
    {
        _cm.Print();
    }
    void PushTack(int taskcode)
    {
        // 1.选择一个子进程，采用轮询，防止负载均衡和负载不均衡
        auto &c = _cm.Select();
        cout << "选择一个子进程：" << c.Name() << endl;
        // 2.发送任务
        c.Send(taskcode);
        cout << "发送了一个任务码：" << taskcode << endl;
    }

private:
    ChannelManager _cm; // 进程链
    int _process_num;   // 进程个数
};

#endif
//Main.cc

#include"ProcessPool.hpp"

int main()
{
    ProcessPool pp(gdefaultnum);
    //创建进程池
    pp.Create();
    //打印进程池
    //pp.Debug();
    int task_code = 1;
   while(true)
   {
    pp.PushTack(task_code++);
    sleep(1);
   }
    return 0;
}

创建进程池后，OS关闭了没有意义的管道，每次选择一个管道接受消息 

原文地址:https://blog.csdn.net/2302_81171591/article/details/146265691
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.kler.cn/a/586612.html

相关文章：

  • 【蓝桥杯】雪地工程核弹引爆控制器最小数量计算
  • Pytorch实现之最小二乘梯度归一化设计
  • 在离线情况下如何使用 Python 翻译文本
  • 【JVM】性能监控与调优概述篇
  • 基于RAGFlow本地部署DeepSpeek-R1大模型与知识库：从配置到应用的全流程解析
  • Spring Boot 中 BootstrapRegistryInitializer 的作用与示例
  • Ubuntu中为curl和Docker配置代理
  • Docker安装GitLab中文版详细流程，以及非80端口配置
  • 2025年渗透测试面试题总结-安恒 （题目+回答）
  • U盘提示格式化的深度解析与应对策略
  • 【新能源汽车研发测试能力建设深度解析：设备、趋势与行业展望】
  • 学习用WinDbg查看程序当前运行的堆栈
  • [C语言日寄] qsort函数的练习
  • css模拟雷达扫描动画
  • 用ST7789屏幕导致负片（反色）的问题
  • Alembic 实战指南：快速入门到FastAPI 集成
  • 深入解析对象存储及工作原理
  • Java 综合实战项目：生成不重复随机字符串数组
  • Android LeakCanary 使用 · 原理详解
  • 微信小程序面试内容整理-数据绑定