进程间通信概述
进程间通信的目的是为了实现以下几个主要功能：
- 数据传输：一个进程需要将其数据发送给另一个进程。
- 资源共享：多个进程之间共享相同的资源。
- 通知事件：一个进程需要向另一个或一组进程发送消息，通知它们发生了某种事件（例如，进程终止时需要通知父进程）。
- 进程控制：某些进程希望完全控制另一个进程的执行（例如，Debug进程），此时控制进程希望能够拦截另一个进程的所有陷入和异常，并能及时了解其状态变化。
进程间通信的发展包括：
- 管道
- System V进程间通信
- POSIX进程间通信
进程间通信的分类包括：
- 管道：
- 匿名管道（pipe）
- 命名管道
- System V IPC：
- 消息队列
- 共享内存
- 信号量
- POSIX IPC：
- 消息队列
- 共享内存
- 信号量
- 互斥量
- 条件变量
- 读写锁
管道
管道是一种最基本的进程间通信机制，主要用于具有亲缘关系的进程之间的通信，例如父子进程。
匿名管道
匿名管道用于父子进程之间的通信。以下是匿名管道的使用示例：
以下是使用管道进行通信的示例代码：
完整代码如下：
#include <iostream> #include <string> #include <cerrno> // errno.h #include <cstring> // string.h #include <unistd.h> #include <sys> #include <sys> const int size = 1024; std::string getOtherMessage(){ static int cnt = 0; std::string messageid = std::to_string(cnt); cnt++; pid_t self_id = getpid(); std::string stringpid = std::to_string(self_id); std::string message = "messageid: "; message += messageid; message += " my pid is : "; message += stringpid; return message; } // 子进程进行写入 void SubProcessWrite(int wfd){ int pipesize = 0; std::string message = "father, I am your son prcess!"; char c = 'A'; while (true) { std::cerr 0) { inbuffer[n] = 0; // == ' ' std::cout 0) exit(0); SubProcessWrite(pipefd[1]); close(pipefd[1]); exit(0); } std::cout 0) { std::cout >8)&0xFF) </sys></sys></unistd.h></cstring></cerrno></string></iostream>
管道的读取和写入行为如下：
- 如果管道内部为空且写端未关闭，读进程将被阻塞，直到写端写入数据。
- 如果管道已满且读端未读取且未关闭，写进程将被阻塞，直到数据被读取。
- 如果管道一直被读且写端关闭，读端的read返回值将为0，表示读到文件结尾。
- 如果读端关闭，写端一直在写入，写端进程将被操作系统用13号信号终止，相当于进程出现异常。
管道的特征
- 匿名管道：仅用于具有亲缘关系的进程之间的通信，常用于父子进程之间。
- 生命周期：随进程而存在。
- 同步机制：管道内部自带进程间同步机制，具有明显的顺序性。
- 字节流：管道文件在通信时是面向字节流的，写的次数和读取的次数不一定一一匹配。
- 通信模式：管道的通信模式是一种特殊的半双工模式，数据只能向一个方向流动；需要双向通信时，需要建立两个管道。
当要写入的数据量不大于PIPE_BUF时，linux将保证写入的原子性；当要写入的数据量大于PIPE_BUF时，Linux不再保证写入的原子性。原子操作意味着写入操作不会被中断，读方要么读不到数据，要么读到完整的数据。
进程池实现
进程池的实现可以利用管道进行进程间通信。以下是进程池实现的示例代码：
ProcessPool.cc：
#include <iostream> #include <string> #include <vector> #include <unistd.h> #include <sys> #include <sys> #include "Task.hpp" // void work(int rfd) // { // while (true) // { // int command = 0; // int n = read(rfd, &command, sizeof(command)); // if (n == sizeof(int)) // { // std::cout 0) // { // std::cout *channels, task_t task){ for (int i = 0; i empty()) { // 第二次之后，开始创建的管道要关闭继承下来的写端 for(auto &channel : *channels) channel.CloseChannel(); } // child - read close(pipefd[1]); dup2(pipefd[0], 0); // 将管道的读端，重定向到标准输入 task(); close(pipefd[0]); exit(0); } // 3.构建一个channel名称 std::string channel_name = "Channel-" + std::to_string(i); // 父进程 close(pipefd[0]); // a. 子进程的pid b. 父进程关心的管道的w端 channels->push_back(Channel(pipefd[1], id, channel_name)); } } } // 0 1 2 3 4 channelnum int NextChannel(int channelnum){ static int next = 0; int channel = next; next++; next %= channelnum; return channel; } void SendTaskCommand(Channel &channel, int taskcommand){ write(channel.GetWfd(), &taskcommand, sizeof(taskcommand)); } void ctrlProcessOnce(std::vector<channel> &channels){ sleep(1); // a. 选择一个任务 int taskcommand = SelectTask(); // b. 选择一个信道和进程 int channel_index = NextChannel(channels.size()); // c. 发送任务 SendTaskCommand(channels[channel_index], taskcommand); std::cout &channels, int times = -1){ if (times > 0) { while (times--) { ctrlProcessOnce(channels); } } else { while (true) { ctrlProcessOnce(channels); } } } void CleanUpChannel(std::vector<channel> &channels){ // int num = channels.size() -1; // while(num >= 0) // { // channels[num].CloseChannel(); // channels[num--].Wait(); // } for (auto &channel : channels) { channel.CloseChannel(); channel.Wait(); } // // 注意 // for (auto &channel : channels) // { // channel.Wait(); // } } // ./processpool 5 int main(int argc, char *argv[]){ if (argc != 2) { std::cerr channels; // 1. 创建信道和子进程 CreateChannelAndSub(num, &channels, work1); // 2. 通过channel控制子进程 ctrlProcess(channels, 5); // 3. 回收管道和子进程. a. 关闭所有的写端 b. 回收子进程 CleanUpChannel(channels); // sleep(100); return 0; }</channel></channel></sys></sys></unistd.h></vector></string></iostream>
Task.hpp：
#pragma once #include <iostream> #include <ctime> #include <cstdlib> #include <sys> #include <unistd.h> #define TaskNum 3 typedef void (*task_t)(); // task_t 函数指针类型 void Print(){ std::cout 2) return; tasks[number](); } int SelectTask(){ return rand() % TaskNum; } void work(){ while (true) { int command = 0; int n = read(0, &command, sizeof(command)); if (n == sizeof(int)) { std::cout </unistd.h></sys></cstdlib></ctime></iostream>
© 版权声明
文章版权归作者所有，未经允许请勿转载。
THE END
