进程间通信方式 - YHT's Note Repo

%% # 纲要 > 主干纲要、Hint/线索/路标 # Q&A #### 已明确 #### 待明确 > 当下仍存有的疑惑 **❓ 有什么问题？** # Buffer ## 闪念 > sudden idea ## 候选资料 > Read it later %% # 进程间的通信方式 ⭐ > **进程间通信**（Inter-Process Communication，IPC）进程间通信方式包括[^1] [^2]： - **管道**（匿名管道 pipe、具名管道 FIFO） - **信号**（异步通知） - **系统 IPC 三件套**（消息队列、信号量、共享内存） - **套接字 Socket** #### IPC 对象的持久性持久性是指 **IPC 对象的生命周期**： - **==进程持久性==**：IPC 对象存在直至 "**最后一个持有该对象的进程关闭**"。 - 包括：管道、FIFO、socket、匿名内存映射 - **==内核持久性==**：IPC 对象**由内核维护**，直至被显式删除或系统关闭时才销毁，与进程是否已打开该对象无关。 - 包括：POSIX IPC 三件套、System IPC 三件套； - **==文件系统持久性==**：**持久存在于文件系统**，直至被从文件系统中删除。 - 唯一具有文件系统持久性的 IPC 对象是 "**基于内存映射文件的共享内存**"。 > [!NOTE] 三种持久性的 IPC 对象 [^2] > > ![[_attachment/02-开发笔记/05-操作系统/进程管理/进程间通信方式.assets/IMG-进程间通信方式-866DBB0E895576B65DB9413A29D9D19F.png|865]] > > [!summary] 总结：各种 IPC 对象的持久性[^3] > > ![[_attachment/02-开发笔记/05-操作系统/进程管理/进程间通信方式.assets/IMG-进程间通信方式-FF1E9BEBA2CF6CB673947049A860B952.png|705]] > ## "通信" 与 "同步" 总结二者侧重点不同： - 通信：关注于 **进程间的数据交换**。 - 同步：关注于 **进程、线程**操作之间的 "互斥"，协调**进程/线程间的执行顺序**，防止产生条件竞争、数据不一致问题。 > [!NOTE] UNIX IPC 涉及的 "**通信**" 与 "**同步**" 工具总览[^1] > > ![[_attachment/02-开发笔记/05-操作系统/进程管理/进程间通信方式.assets/IMG-进程间通信方式-F386BC1A0E0333527577074F69986048.png|569]] > > 关于上图中 "同步" 工具： > > - 条件变量、互斥锁通常用于 "**线程间同步**"，也可用于 "**进程间同步**"（要求将条件变量、互斥锁以 "**共享内存**" 的方式供**跨进程访问**） > - 信号量即可用于 "**进程间同步**"，也可用于 "**线程间同步**"。 > > 关于系统 IPC 三件套（消息队列、信号量、共享内存）： > > - "**POSIX IPC 三件套**" 是对早期 **"SystemV IPC 三件套"** 实现的改进。 > [!summary] 总结：进程、线程间 "通信" 与 "同步" 方式 > > - **进程间通信**（消息传递） > - **管道**（匿名管道 pipe、具名管道 FIFO） > - **信号**（异步通知） > - **消息队列** > - **共享内存** > - **套接字 Socket** > > - **进程间同步** > - **信号量** > - **文件锁** > - **互斥锁 & 条件变量**（需作为 "**共享内存**" 访问） > > - 线程间通信 ❌（不需要，多线程共享进程的内存空间，在同步机制下安全访问进程内的全局变量即可） > > - **线程间同步**： > - 互斥锁 > - 条件变量 > - 信号量 > - 屏障 > - 原子操作 > # 管道「**管道**」是本质上是 **==内核中的一段缓冲区==**，从管道一端写入的数据**缓存在内核中**，而由另一端进行读取。特点： - 典型实现为 "**半双工通信**"：数据只能单向流动； - 通过 **文件 I/O** 直接读/写管道（`read`, `write` 等） - **管道中的数据遵循 FIFO 原则**，不支持 lseek 之类的文件定位。 > [!NOTE] POSIX 标准库提供的管道 API： > > - **匿名管道**：`pipe`、`popen`、`pclose` > - **具名管道**（FIFO）：`mkfifo`、`mkfifoat` > > ### 管道的类型两种类型： - **==匿名管道==**：**通过 `pipe()` 创建**，只在进程中有效，故只能在 **==具有公共祖先的两个关联进程之间==** 使用； - **==具名管道==**（FIFO）：**通过 `mkfifo()` 创建**，会在文件系统中创建一个 FIFO 文件，因此 **==可用于无关联的任意进程间的通信==**。 > [!NOTE] 管道的生命周期 > > - **匿名管道**：随进程，当 **"最后一个" 引用管道的进程终止时**，管道也被完全地删除。 > - **具名管道 FIFO**： > - 当最后一个引用 FIFO 的进程终止时，**FIFO 中的数据会被删除**。 > - **FIFO 文件**会**持久化**在文件系统中，**直至被显式删除**。 > > [!example] 匿名管道的典型用法——**父子进程间的单向通信** > > 典型用法： > > 1. 父进程创建管道，并 `fork` 生成子进程； > 2. 关闭其中一方的 "**管道读端**"，同时关闭另一方的 "**管道写端**" > > 由此，构成了父子进程间的 "**单向通信**"，**两个进程分别负责管道的读或写**。 > > > ![[_attachment/02-开发笔记/05-操作系统/进程管理/进程间通信方式.assets/IMG-进程间通信方式-7C2131A47DBA02A42406DAF2D1E4368D.png|437]] > > [!example] shell 中的 "管道命令 `|`" 背后即是创建 "匿名管道" > > ![[_attachment/02-开发笔记/05-操作系统/进程管理/进程间通信方式.assets/IMG-进程间通信方式-6804931E1BEDF33F291226A7C0F7D214.png|665]] > ## 匿名管道 ### pipe 函数 ![[_attachment/02-开发笔记/05-操作系统/进程管理/进程间通信方式.assets/IMG-进程间通信方式-C15E53E019ED83A8C5D79225A80BE96A.png|647]] `pipe` 用于创建 **==匿名管道==**，其存在于 "**内存中的内核地址空间**"。创建成功时，传入参数 `fd` 将被赋予两个 "**文件描述符**"，提供一个 "**==单向数据流==**"： - `fd[0]` 对应**管道的读端**； - `fd[1]` 对应**管道的写端**，其**输出将是 `fd[0]`的输入**。 > [!NOTE] `pipe()` 系统调用打开的两个文件描述符是 "**内核管道**" 的读写端。 > ![[_attachment/02-开发笔记/05-操作系统/进程管理/进程间通信方式.assets/IMG-进程间通信方式-F491CDA8FDC7519D55370A1F8AAC15CB.png|398]] > [!NOTE] 实现 "**双向通信**" 需要打开两个管道 > > ![[_attachment/02-开发笔记/05-操作系统/进程管理/进程间通信方式.assets/IMG-进程间通信方式-E98D652BCE03F4B9FC9309C171011B27.png|406]] > ### popen 与 pclose 函数 > 位于头文件 `<stdio.h>` 中 > ![[_attachment/02-开发笔记/05-操作系统/进程管理/进程间通信方式.assets/IMG-进程间通信方式-7D2C8BF619B39CBE962E0ACCEE82B63B.png|732]] `popen` 创建一个管道并 **==启动另一个进程==**，返回一个**标准 I/O 文件指针**（根据参数 `type` 类型，指针指向不同）： - 若 `type == 'r'`，则指针指向 `command` 进程的 **标准输出**，可读取来自 `command` 的输出； - 若 `type == 'w'`，则指针指向 `command` 进程的 **标准输入**，可写入 `command` 的标准输入； `pclose` 关闭标准 I/O 流，**阻塞等待 `popen` 创建的子进程结束**。 > [!NOTE] `popen` 是对 `pipe` 、`fork` 、`exec`的封装，其返回一个标准 I/O 文件指针，指向所创建子进程的标准输入或标准输出。 ### 匿名管道使用示例 ![[_attachment/02-开发笔记/05-操作系统/进程管理/进程间通信方式.assets/IMG-进程间通信方式-CFB14C38D7D71CA008E6B64129B17AA6.png|537]] 下例中，父子进程之间的通信示意图如上： ```c #include <stdio.h> #include <fcntl.h> #include <unistd.h> #include <stdlib.h> #include <sys/wait.h> int main(int argc, char** argv) { int fd[2]; // 创建一个管道, 获取两个文件描述符fd[0]与fd[1], 后者用于写, 前者用于读, 后者的输出作为前者的输入 if (pipe(fd) == -1) { printf("pipe error\n"); exit(1); } // 创建第一个子进程, 将其标准输入写入到fd[1] pid_t pid1 = fork(); if (pid1 < 0) { printf("fork failed\n"); exit(1); } else if (pid1 == 0) { // 子进程1 close(fd[0]); // 关闭读端 dup2(fd[1], STDOUT_FILENO); // 将标准输出重定向到管道的写端. close(fd[1]); // 关闭写端(已完成重定向, 此后fd[1]已无用) printf("hello, world!\n"); // 子进程1标准输出=>重定向到管道写端. exit(0); } // 创建第二个子进程, 读取fd[0]. pid_t pid2 = fork(); if (pid2 < 0) { printf("fork failed\n"); } else if (pid2 == 0) { close(fd[1]); // 关闭写端; dup2(fd[0], STDIN_FILENO); // 将标准输入重定向到管道的读端 close(fd[0]); // 关闭读端(已完成重定向, 此后不需要再使用fd[0]). // 子进程2执行cat命令, 其标准输入已重定向到管道读端. char* args[] = {"cat", NULL}; execvp("cat", args); perror("execvp failed"); exit(0); } // 父进程: 关闭管道的读写端, 等待两个子进程结束. close(fd[0]); close(fd[1]); // 等待两个子进程结束 waitpid(pid1, NULL, 0); waitpid(pid2, NULL, 0); return 0; } ``` ## 具名管道 > ![[_attachment/02-开发笔记/05-操作系统/进程管理/进程间通信方式.assets/IMG-进程间通信方式-B4CAE0AE239FF08E03A8465BD6569DEF.png|647]] `mkfifo` 或 `mkfifoat` **创建或打开一个已存在的具名管道**，**进程间通过文件 I/O 函数**来读写 FIFO 文件，**实现通信**。 > [!NOTE] linux 中具有**同名的 `mkfifo` 命令** # 消息队列「**消息队列**」：由操作系统**内核**维护的**消息链接表**，由**消息队列标识符**标识，用于**存储消息**，实现**进程间==异步==通信**。 ### 消息队列的特点特点： - **异步通信**：消息队列不需要发送方和接收方同时工作，消息**保存在队列中，直到接收方准备好读取** - **类型化消息**：每个消息具有 "**消息类型**" 和 "**消息内容**" 两项信息，其中消息类型由 "**类型标识符**" 指示，**进程可筛选和读取特定类型消息** - **持久化消息**：消息队列**生命周期随内核**，**消息在队列中存在==直到被读取或显式删除==**，因此消息不会因进程的退出而丢失。 - **可按优先级排序**：默认采用 FIFO 队列，也可根据 "**消息优先级**" 改为 "**优先队列**"。缺点： - **通信不及时** - **不适合大数据传输**：内核中**每个消息体都有一个最大长度的限制**，同时**所有队列所包含的全部消息体的总长度也是有上限** - Linux 内核中，宏定义 `MSGMAX` 和 `MSGMNB`，以字节为单位，分别**定义了==一条消息的最大长度==和==一个队列的最大长度==**。 - 向**消息队列写入、读取消息**的过程中存在 "**用户态-内核态**" 之间的切换开销 ### 消息队列工作过程 - **消息发送**：进程通过系统调用将一条消息发送到消息队列中。消息包含消息类型（标识符）和消息数据。 - **消息存储**：消息在队列中按照 **FIFO 顺序**或**优先级顺序**（优先队列）存储，直到接收方读取。 - **消息读取**：接收进程通过系统调用读取队列中的消息。它可以选择读取特定类型的消息，或者按队列的顺序读取消息。消息队列中每条消息以 "**==消息体==**" （用户自定义数据类型）形式作为**数据单元**，具有**固定大小**，由**消息发送方和接收方约定好**。 ## OS 消息队列接口消息队列有不同标准下的实现： - **POSIX 消息队列**： - POSIX 标准提供了 `mq_open`、`mq_send`、`mq_receive` 等 API 来管理消息队列，位于头文件 `<mqueue.h>` 中 - **System V 消息队列**： - System V 消息队列（早期的 IPC 机制）通过 `msgget`、`msgsnd` 和 `msgrcv` 函数来操作消息队列，位于头文件 `<sys/msg.h>` 中 ![[_attachment/02-开发笔记/05-操作系统/进程管理/进程间通信方式.assets/IMG-进程间通信方式-A786F62A918B93FA3565E1D1F54ABB89.png|494]] #### POSIX 消息队列 ```c #include <stdio.h> #include <stdlib.h> #include <fcntl.h> /* FOr O_* constants */ #include <mqueue.h> /* For POSIX message queues */ #include <sys/stat.h> /* For mode constants */ #include <string.h> int main() { mqd_t mq; // 消息队列描述符 struct mq_attr attr; // mqueue中提供的消息队列属性 const char* message = "Hello, POSIX Message queue!"; char buffer[1024]; // 设置消息队列属性 attr.mq_flags = 0; // 阻塞模式 attr.mq_maxmsg = 10; // 最大消息数 attr.mq_msgsize = 1024; // 消息最大大小 attr.mq_curmsgs = 0; // 当前消息数量 // 创建消息队列(为什么是创建文件?) mq = mq_open("/my_queue", O_CREAT | O_RDWR, 0644, &attr); if (mq == (mqd_t) - 1) { perror("mq_open"); exit(1); } // 发送消息: mq_send if (mq_send(mq, message, strlen(message) + 1, 0) == -1) { perror("mq_send"); exit(1); } printf("消息已发送: %s\n", message); // 接收消息: mq_receive if (mq_receive(mq, buffer, 1024, NULL) == -1) { perror("mq_receive"); exit(1); } printf("接收到的消息: %s\n", buffer); // 关闭并删除消息队列 mq_close(mq); mq_unlink("/my_queue"); return 0; } ``` #### System V 消息队列 ```c #include <sys/ipc.h> // System V IPC, for ftok. #include <sys/msg.h> // System V message queue, for msgget, msgsnd, msgrcv, msgctl #include <string.h> #include <stdio.h> #include <sys/wait.h> #include <unistd.h> #include <stdlib.h> #include <assert.h> // 定义消息结构体 struct msg_buffer { long msg_type; // 消息类型，必须为long类型 char msg_text[100]; }; int main() { key_t key; int msgid; // 生成一个唯一的key, 用作消息队列IPC对象的外部名, 通常通过文件路径和整数生成 key = ftok("profile", 65); // 创建消息队列, 返回消息队列ID标识符 msgid = msgget(key, 0666 | IPC_CREAT); struct msg_buffer message; // 消息体 message.msg_type = 1; // 设置消息类型 // 创建父子进程, 一个发送一个读取 pid_t pid = fork(); if (pid < 0) { perror("fork failed"); exit(1); } if (pid > 0) { // parent // 准备要发送的消息内容 strcpy(message.msg_text, "Hello, this is a message!"); // 发生消息到消息队列: msgsnd msgsnd(msgid, &message, sizeof(message), 0); printf("发送的消息: %s\n", message.msg_text); } else { // child // 接收消息 msgrcv(msgid, &message, sizeof(message), 1, 0); // 接收消息类型为1 printf("接收到的消息: %s\n", message.msg_text); } int rc = wait(NULL); assert(rc == 0); msgctl(msgid, IPC_RMID, NULL); return 0; } ``` # 共享内存参见[[02-开发笔记/05-操作系统/进程管理/共享内存|共享内存]] # 信号量 - 概念参见：[[02-开发笔记/05-操作系统/并发相关/同步原语#信号量|同步原语#信号量]] - 接口参见：[[02-开发笔记/05-操作系统/并发相关/POSIX 线程&同步 API|POSIX 线程&同步 API]] --- # 参考资料 # Footnotes [^1]: 《Linux-UNIX 系统编程手册-下册》 P719 [^2]: 《UNIX 网络编程卷 2：进程间通信》 P4 [^3]: 《Linux-UNIX 系统编程手册-下册》 P726