devbox
花生_TL007
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Linux进程间通信（管道）_linux 管道

作者：花生_TL007 | 2024-03-08 20:50:55

赞

踩

linux 管道

目录

进程间通信的目的

进程间通信的必要性本质以及技术背景

管道 

匿名管道 

扩展 

命名管道

进程间通信的目的

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

进程间通信的必要性本质以及技术背景

进程间通信的必要性，单进程的，那么也就无法使用并发能力，更加无法实现多进程协同。传输数据，同步执行流，消息通知等，所以进程通信不是目的，而是手段。

进程间通信的技术背景

1．进程是具有独立性的。虚拟地址空间+页表保证进程运行的独立性(进程内核数据结构+进程的代码和数据)。

2．通信成本会比较高。

进程间通信的本质理解

1．进程间通信的前提，首先需要让不同的进程看到同一块"内存"(特定的结构组织的)

2．所以你所谓的进程看到同一块"内存"，属于哪一个进程呢?不能隶属于任何一个进程，而应该更强调共享。 

管道 

管道是Unix中最古老的进程间通信的形式。
我们把从一个进程连接到另一个进程的一个数据流称为一个“管道”
管道又分为两种，匿名管道和命名管道。

管道的原理：

这个就叫做管道，分别以读写的方式打开一个文件，fork()创建子进程，双方关闭自己不需要的文件描述符，管道就是文件，两个进程通过文件的方式进行通信。

匿名管道 

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


其实从上面这张都就可以看出管道是怎们样的，接下来就用代码来实现一下。 

  1. #include <iostream>
  2. #include <string>
  3. #include <cstdio>
  4. #include <cstring>
  5. #include <assert.h>
  6. #include <unistd.h>
  7. #include <sys/types.h>
  8. #include <sys/wait.h>
  9.  
  10. using namespace std;
  11.  
  12. // 为什么不定义全局buffer来进行通信呢？？ 因为有写时拷贝的存在，无法更改通信！
  13.  
  14. int main()
  15. {
  16.     // 1. 创建管道
  17.     int pipefd[2] = {0};
  18.     int n = pipe(pipefd);
  19.     assert(n == 0);
  20.     (void)n;
  21. #ifdef DEBUG
  22.     cout << "pipefd[0]: " << pipefd[0] << endl; // 3
  23.     cout << "pipefd[1]: " << pipefd[1] << endl; // 4
  24. #endif
  25.     // 2. 创建子进程
  26.     pid_t id = fork();
  27.     assert(id != -1);
  28.     if (id == 0)
  29.     {
  30.         // 子进程 - 读
  31.         // 3. 构建单向通信的信道，父进程写入，子进程读取
  32.         // 3.1 关闭子进程不需要的fd
  33.         close(pipefd[1]);
  34.         char buffer[1024 * 8];
  35.         while (true)
  36.         {
  37.             // sleep(20);
  38.             // 写入的一方，fd没有关闭，如果有数据，就读，没有数据就等
  39.             // 写入的一方，fd关闭, 读取的一方，read会返回0，表示读到了文件的结尾！
  40.             ssize_t s = read(pipefd[0], buffer, sizeof(buffer) - 1);
  41.             if (s > 0)
  42.             {
  43.                 buffer[s] = '\0';
  44.                 cout << "child get a message[" << getpid() << "] Father# " << buffer << endl;
  45.             }
  46.             else if (s == 0)
  47.             {
  48.                 cout << "writer quit(father), me quit!!!" << endl;
  49.                 break;
  50.             }
  51.         }
  52.         exit(1);
  53.     }
  54.     // 父进程 - 写
  55.     //  3. 构建单向通信的信道
  56.     //  3.1 关闭父进程不需要的fd
  57.     close(pipefd[0]);
  58.     string message = "我是父进程，我正在给你发消息";
  59.     char buffer[1024 * 8];
  60.     int count = 0;
  61.     char send_buffer[1024 * 8];
  62.     while (true)
  63.     {
  64.         // 3.2 构建一个变化的字符串
  65.         snprintf(send_buffer, sizeof(send_buffer), "%s[%d] : %d",
  66.                  message.c_str(), getpid(), count++);
  67.         // 3.3 写入
  68.         write(pipefd[1], send_buffer, strlen(send_buffer));
  69.         // 3.4 故意sleep
  70.         sleep(1);
  71.         cout << count << endl;
  72.         if (count == 5)
  73.         {
  74.             cout << "writer quit(father)" << endl;
  75.             break;
  76.         }
  77.     }
  78.     close(pipefd[1]);
  79.     pid_t ret = waitpid(id, nullptr, 0);
  80.     cout << "id : " << id << " ret: " << ret << endl;
  81.     assert(ret > 0);
  82.     (void)ret;
  83.  
  84.     return 0;
  85. }

来讲一下这个代码的大致方向，第一部肯定是创建管道，然后就是创建子进程，这里我们是让父进程写子进程读，所以关闭父子进程不需要的fd，写入的一方，fd没有关闭，如果有数据，就读，没有数据就等，写入的一方，fd关闭, 读取的一方，read会返回0，表示读到了文件的结尾，最后关闭父子进程fd，回收子进程。

其从上面的代码我们就可以总结一下管道的特点。总结管道的特点，理解以前的管道 | ，管道是一个文件–读取–具有访问控制，显示器也是一个文件，父子同时往显示器写入的时候，有没有说一个会等另一个的情况呢，缺乏访问控制。

1．管道是用来进行具有血缘关系的进程进性进程间通信-- 常用于父子通信 

2.管道具有通过让进程间协同，提供了访问控制

3.管道提供的是面向流式的通信服务--面向字节流--_协议 

4．管道是基于文件的，文件的生命周期是随进程的，管道的生命周期是随进程的

5.管道是单向通信的，就是半双工通信的一种特殊情况 

6.写快，读慢，写满不能在写了

7.写慢，读快，管道没有数据的时候，读必须等待

8.写关，读0，标识读到了文件结尾

9.读关，写继续写，oS终止写进程

扩展 

分发多个任务的管理器

 代码：

  1. //.hpp
  2. //#pragma once
  3.  
  4. #include <iostream>
  5. #include <string>
  6. #include <vector>
  7. #include <unordered_map>
  8. #include <unistd.h>
  9. #include <functional>
  10.  
  11. typedef std::function<void()> func;
  12.  
  13. std::vector<func> callbacks;
  14. std::unordered_map<int, std::string> desc;
  15.  
  16. void readMySQL()
  17. {
  18.     std::cout << "sub process[" << getpid() << " ] 执行访问数据库的任务\n" << std::endl;
  19. }
  20.  
  21. void execuleUrl()
  22. {
  23.     std::cout << "sub process[" << getpid() << " ] 执行url解析\n" << std::endl;
  24. }
  25.  
  26. void cal()
  27. {
  28.     std::cout << "sub process[" << getpid() << " ] 执行加密任务\n" << std::endl;
  29. }
  30.  
  31. void save()
  32. {
  33.     std::cout << "sub process[" << getpid() << " ] 执行数据持久化任务\n" << std::endl;
  34. }
  35.  
  36. void load()
  37. {
  38.     desc.insert({callbacks.size(), "readMySQL: 读取数据库"});
  39.     callbacks.push_back(readMySQL);
  40.  
  41.     desc.insert({callbacks.size(), "execuleUrl: 进行url解析"});
  42.     callbacks.push_back(execuleUrl);
  43.  
  44.     desc.insert({callbacks.size(), "cal: 进行加密计算"});
  45.     callbacks.push_back(cal);
  46.  
  47.     desc.insert({callbacks.size(), "save: 进行数据的文件保存"});
  48.     callbacks.push_back(save);
  49. }
  50.  
  51. void showHandler()
  52. {
  53.     for(const auto &iter : desc )
  54.     {
  55.         std::cout << iter.first << "\t" << iter.second << std::endl;
  56.     }
  57. }
  58.  
  59. int handlerSize()
  60. {
  61.     return callbacks.size();
  62. }
  63. //.cc
  64. #include <iostream>
  65. #include <vector>
  66. #include <cstdlib>
  67. #include <ctime>
  68. #include <cassert>
  69. #include <unistd.h>
  70. #include <sys/wait.h>
  71. #include <sys/types.h>
  72. #include "Task.hpp"
  73.  
  74. #define PROCESS_NUM 5
  75.  
  76. using namespace std;
  77.  
  78. int waitCommand(int waitFd, bool &quit) // 如果对方不发，我们就阻塞
  79. {
  80.     uint32_t command = 0;
  81.     ssize_t s = read(waitFd, &command, sizeof(command));
  82.     if (s == 0)
  83.     {
  84.         quit = true;
  85.         return -1;
  86.     }
  87.     assert(s == sizeof(uint32_t));
  88.     return command;
  89. }
  90.  
  91. void sendAndWakeup(pid_t who, int fd, uint32_t command)
  92. {
  93.     write(fd, &command, sizeof(command));
  94.     cout << "main process: call process " << who << " execute " << desc[command] << " through " << fd << endl;
  95. }
  96.  
  97. int main()
  98. {
  99.     // 代码中关于fd的处理，有一个小问题，不影响我们使用，但是你能找到吗？？
  100.     load();
  101.     // pid: pipefd
  102.     vector<pair<pid_t, int>> slots;
  103.     // 先创建多个进程
  104.     for (int i = 0; i < PROCESS_NUM; i++)
  105.     {
  106.         int pipefd[2] = {0};
  107.         int n = pipe(pipefd);
  108.         assert(n == 0);
  109.         (void)n;
  110.         pid_t id = fork();
  111.         assert(id != -1);
  112.         if (id == 0)
  113.         {
  114.             close(pipefd[1]);
  115.             while (true)
  116.             {
  117.                 // pipefd[0]
  118.                 // 等命令
  119.                 bool quit = false;
  120.                 int command = waitCommand(pipefd[0], quit); // 如果对方不发，我们就阻塞
  121.                 if (quit)
  122.                     break;
  123.                 // 执行对应的命令
  124.                 if (command >= 0 && command < handlerSize())
  125.                 {
  126.                     callbacks[command]();
  127.                 }
  128.                 else
  129.                 {
  130.                     cout << "非法command: " << command << endl;
  131.                 }
  132.             }
  133.             exit(0);
  134.         }
  135.         // father,进行写入，关闭读端
  136.         close(pipefd[0]); // pipefd[1]
  137.         slots.push_back(pair<pid_t, int>(id, pipefd[1]));
  138.     }
  139.     // 父进程派发任务
  140.     srand((unsigned long)time(nullptr) ^ getpid() ^ 23323123123L); // 让数据源更随机
  141.     while (true)
  142.     {
  143.         // 选择一个任务, 如果任务是从网络里面来的？
  144.         int command = rand() % handlerSize();
  145.         // 选择一个进程 ，采用随机数的方式，选择进程来完成任务，随机数方式的负载均衡
  146.         int choice = rand() % slots.size();
  147.         // 把任务给指定的进程
  148.         sendAndWakeup(slots[choice].first, slots[choice].second, command);
  149.         sleep(1);
  150.         // int select;
  151.         // int command;
  152.         // cout << "############################################" << endl;
  153.         // cout << "#   1. show funcitons      2.send command  #" << endl;
  154.         // cout << "############################################" << endl;
  155.         // cout << "Please Select> ";
  156.         // cin >> select;
  157.         // if (select == 1)
  158.         //     showHandler();
  159.         // else if (select == 2)
  160.         // {
  161.         //     cout << "Enter Your Command> ";
  162.         //     // 选择任务
  163.         //     cin >> command;
  164.         //     // 选择进程
  165.         //     int choice = rand() % slots.size();
  166.         //     // 把任务给指定的进程
  167.         //     sendAndWakeup(slots[choice].first, slots[choice].second, command);
  168.         // }
  169.         // else
  170.         // {
  171.         // }
  172.     }
  173.     // 关闭fd, 所有的子进程都会退出
  174.     for (const auto &slot : slots)
  175.     {
  176.         close(slot.second);
  177.     }
  178.  
  179.     // 回收所有的子进程信息
  180.     for (const auto &slot : slots)
  181.     {
  182.         waitpid(slot.first, nullptr, 0);
  183.     }
  184. }

命名管道

 命名管道可以从命令行上创建，命令行方法是使用下面这个命令 mkfifo filename                         删除管道文件：unlink filename
命名管道也可以从程序里创建，相关函数是：

匿名管道与命名管道的区别
匿名管道由pipe函数创建并打开。
命名管道由mkfifo函数创建，打开用open
FIFO（命名管道）与pipe（匿名管道）之间唯一的区别在它们创建与打开的方式不同，一但这些工作完成之后，它们具有相同的语义
接下来同样用一段代码来演示一下:

  1. //Makefile
  2. .PHONY:all
  3. all:client mutiServer
  4.  
  5. client:client.cxx
  6. 	g++ -o $@ $^ -std=c++11
  7. mutiServer:server.cxx
  8. 	g++ -o $@ $^ -std=c++11
  9.  
  10. .PHONY:clean
  11. clean:
  12. 	rm -f client mutiServer
  13. //client.cxx
  14. #include "comm.hpp"
  15.  
  16. int main()
  17. {
  18.     // 1. 获取管道文件
  19.     int fd = open(ipcPath.c_str(), O_WRONLY);
  20.     if(fd < 0)
  21.     {
  22.         perror("open");
  23.         exit(1);
  24.     }
  25.  
  26.     // 2. ipc过程
  27.     string buffer;
  28.     while(true)
  29.     {
  30.         cout << "Please Enter Message Line :> ";
  31.         std::getline(std::cin, buffer);
  32.         write(fd, buffer.c_str(), buffer.size());
  33.     }
  34.  
  35.     // 3. 关闭
  36.     close(fd);
  37.     return 0;
  38. }
  39. //log.hpp
  40. #ifndef _LOG_H_
  41. #define _LOG_H_
  42.  
  43. #include <iostream>
  44. #include <ctime>
  45.  
  46. #define Debug   0
  47. #define Notice  1
  48. #define Warning 2
  49. #define Error   3
  50.  
  51.  
  52. const std::string msg[] = {
  53.     "Debug",
  54.     "Notice",
  55.     "Warning",
  56.     "Error"
  57. };
  58.  
  59. std::ostream &Log(std::string message, int level)
  60. {
  61.     std::cout << " | " << (unsigned)time(nullptr) << " | " << msg[level] << " | " << message;
  62.     return std::cout;
  63. }
  64.  
  65.  
  66. #endif
  67. //comm.hpp
  68. #ifndef _COMM_H_
  69. #define _COMM_H_
  70.  
  71. #include <iostream>
  72. #include <string>
  73. #include <cstdio>
  74. #include <cstring>
  75. #include <unistd.h>
  76. #include <sys/types.h>
  77. #include <sys/stat.h>
  78. #include <fcntl.h>
  79. #include "Log.hpp"
  80.  
  81. using namespace std;
  82.  
  83. #define MODE 0666
  84. #define SIZE 128
  85.  
  86. string ipcPath = "./fifo.ipc";
  87.  
  88.  
  89. #endif
  90. //server.cxx
  91. #include "comm.hpp"
  92. #include <sys/wait.h>
  93.  
  94. static void getMessage(int fd)
  95. {
  96.     char buffer[SIZE];
  97.     while (true)
  98.     {
  99.         memset(buffer, '\0', sizeof(buffer));
  100.         ssize_t s = read(fd, buffer, sizeof(buffer) - 1);
  101.         if (s > 0)
  102.         {
  103.             cout <<"["  << getpid() << "] "<< "client say> " << buffer << endl;
  104.         }
  105.         else if (s == 0)
  106.         {
  107.             // end of file
  108.             cerr <<"["  << getpid() << "] " << "read end of file, clien quit, server quit too!" << endl;
  109.             break;
  110.         }
  111.         else
  112.         {
  113.             // read error
  114.             perror("read");
  115.             break;
  116.         }
  117.     }
  118. }
  119.  
  120. int main()
  121. {
  122.     // 1. 创建管道文件
  123.     if (mkfifo(ipcPath.c_str(), MODE) < 0)
  124.     {
  125.         perror("mkfifo");
  126.         exit(1);
  127.     }
  128.  
  129.     Log("创建管道文件成功", Debug) << " step 1" << endl;
  130.  
  131.     // 2. 正常的文件操作
  132.     int fd = open(ipcPath.c_str(), O_RDONLY);
  133.     if (fd < 0)
  134.     {
  135.         perror("open");
  136.         exit(2);
  137.     }
  138.     Log("打开管道文件成功", Debug) << " step 2" << endl;
  139.  
  140.     int nums = 3;
  141.     for (int i = 0; i < nums; i++)
  142.     {
  143.         pid_t id = fork();
  144.         if (id == 0)
  145.         {
  146.             // 3. 编写正常的通信代码了
  147.             getMessage(fd);
  148.             exit(1);
  149.         }
  150.     }
  151.     for(int i = 0; i < nums; i++)
  152.     {
  153.         waitpid(-1, nullptr, 0);
  154.     }
  155.     // 4. 关闭文件
  156.     close(fd);
  157.     Log("关闭管道文件成功", Debug) << " step 3" << endl;
  158.  
  159.     unlink(ipcPath.c_str()); // 通信完毕，就删除文件
  160.     Log("删除管道文件成功", Debug) << " step 4" << endl;
  161.  
  162.     return 0;
  163. }

上面的代码其实就是实现server进程和client进程间的通信，其实从这些代码也可以看出来，匿名管道适用于父子进程间的通信，命名管道用于两个毫不相关的进程间通信。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/花生_TL007/article/detail/212045
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号