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

Linux 进程间通信底层原理(1)：匿名与命令管道

news 2025/11/10 14:05:20

1.进程间通信

进程间通信⽬的：

进程间通信发展：

进程间通信分类：

2.管道

匿名管道：

匿名管道的原理：

匿名管道的创建：

匿名管道的特性：

实战运用：

一、通信管道封装类

二、子进程任务处理函数

三、任务函数定义

四、进程池初始化函数

五、任务分配函数

六、进程池销毁机制

七、主函数入口

命名管道：

命名管道的原理：

命名管道的创建：

实战运用：

匿名管道与命名管道的区别：

1.进程间通信

为什么进程还需要通信呢？在 Linux 中，进程并非孤立存在，它们往往需要通过通信协同完成复杂任务。进程需要通信的核心原因可以概括为：打破进程间的隔离性，实现数据交换、任务协作和资源共享。

进程间通信⽬的：

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

进程间通信发展：

管道
System V进程间通信
POSIX进程间通信

进程间通信分类：

管道： 匿名管道与命名管道
System V IPC：System V 消息队列 System V共享内存 System V 信号量
POSIX IPC：消息队列共享内存信号量互斥量条件变量读写锁

补充：1.两个进程之间需要通信交流，进程间通信的本质，必须让不同的进程看到同一份“资源”，即一份内存空间；2.一般由操作系统来提供这一内存空间，为什么不是其中一个进程提供？会破坏进程的独立性！3.操作系统提供系统调用接口用于通信，这个通信模块属于文件系统，IPC通信模块（基于system V和posix标准）4.而文件之间的通信通常使用管道。

2.管道

何为管道？管道是Unix中最古⽼的进程间通信的形式；我们把从⼀个进程连接到另⼀个进程的⼀个数据流称为⼀个“管道”，管道分为匿名管道和命名管道；

匿名管道：

匿名管道的原理：

当父进程创建子进程时，会写时拷贝PCB，虚拟地址空间这些东西，文件描述符file_struct也会拷贝，那么文件是否会拷贝？(struct file) ，不会而是父子进程共同指向同一个文件
管道就是文件，需要内存，与文件性质一样，只是不刷新到磁盘中去，通信时通过读写两种方式分别打开同一个管道文件，因为打开方式不同，所以会分配两个文件描述符，创建子进程时文件描述符拷贝，file_struct拷贝，但是文件层面相同，并且指向同一块文件缓冲区，根据需要父子进程分别关闭一个读写文件，实现单向通信——管道
如果没有任何关系，那么不能通过这一个原理进行通信，必须有血缘关系，常用于父子进程
建立信道是必然的，成本也高昂，是因为进程具有独立性

匿名管道的创建：

介绍一下系统调用接口：

int pipe(int pipefd[2]);其中为输出型参数，pipefd[0]为读文件描述符，pipefd[1]为写文件描述符

下面给出一张图方便理解：

可见，如果是创建一个匿名管道供父子进程使用的话，确实会因为打开方式不同，所以会分配两个文件描述符，文件层面相同，指向同一块文件缓冲区，根据需要父子进程分别关闭一个读端一个写端，实现单向通信。

匿名管道的特性：

管道的特征：

具有血缘关系之间的进程进行通信
管道只能单向通信，管道是半双工的，数据只能向一个方向流动；需要双方通信时，需要建立起两个管道
父子进程是会进程协同的，同步与互斥——保护管道文件的数据安全
管道是面向字节流的
管道是基于文件的，如果不关闭开启的管道，最后会被操作系统释放掉，其生命周期随进程

管道通信中的四种情况：

读写端正常，管道如果为空，读端就要阻塞
读写端正常，管道如果被写满，写端就要阻塞
读端正常读，写端关闭，读端读到0，表明读到了文件结尾，不会被阻塞
写端正常写，读端关闭，操作系统会通过13号信号SIGPIPE杀掉写端进程

tip：SIGPIPE 是一个信号常量，其值通常为 13，信号是一种软件中断机制，用于通知进程发生了某种特定的事件。SIGPIPE 信号主要与管道（包括匿名管道和命名管道）以及网络套接字通信相关！

关于管道写入的原子性：

PIPE_BUF：原子写入的最大字节数，为4KB，超过这个大小可能导致数据混乱，管道的大小是64KB

当要写⼊的数据量不⼤于PIPE_BUF时，linux将保证写⼊的原⼦性。

当要写⼊的数据量⼤于PIPE_BUF时，linux将不再保证写⼊的原⼦性。

实战运用：

基于匿名管道实现简单版本进程池：基于父进程分配任务给子进程去执行：

总体思路：把父进程创建的所有匿名管道当成一个类进行封装；最后在组织起来再封装成一个管理管道的类；最后就是由进程池这个主类进行控制即可。

下面看粗略理解进程池实现的轮廓；之后辅助代码实现：

一、通信管道封装类

class channel
{
public:// 构造函数：初始化管道写端、进程ID和名称channel(int wfd, pid_t id, const string& name): cwdfd(wfd), slaverid(id), processname(name) {}// 获取管道写端文件描述符int getCwdFd() const { return cwdfd; }// 获取子进程PIDpid_t getSlaverId() const { return slaverid; }// 获取进程名称string getProcessName() const { return processname; }private:int cwdfd;         // 管道写端文件描述符pid_t slaverid;    // 子进程PIDstring processname;// 进程名称（用于日志）
};

二、子进程任务处理函数

// 子进程任务执行循环
void slaver(const vector<function<void()>>& tasks)
{while(1){int cmdcode = 0;// 从标准输入（已重定向为管道读端）读取任务编号ssize_t n = read(0, &cmdcode, sizeof(int)); if(n == sizeof(int)){// 执行对应编号的任务cout << "child says: recieve the message " << cmdcode << " pid: " << getpid() << endl;tasks[cmdcode](); }else if(n == 0){// 读端关闭，退出循环break;}}
}

三、任务函数定义

// 任务1实现
void task1()
{cout << "Executing task 0 in process pool." << endl;
}// 任务2实现
void task2()
{cout << "Executing task 1 in process pool." << endl;
}// 任务3实现
void task3()
{cout << "Executing task 2 in process pool." << endl;
}// 任务4实现
void task4()
{cout << "Executing task 3 in process pool." << endl;
}

四、进程池初始化函数

void initprocesspool(vector<channel>& channels, vector<function<void()>>& tasks)
{// 初始化任务队列tasks = {task1, task2, task3, task4};vector<int> rubbish; // 用于子进程关闭多余管道// 创建指定数量的子进程for(int i = 0; i < processnum; i++){int pipefd[2];// 创建管道int ret = pipe(pipefd);if(ret == -1) perror("pipe error");pid_t pid = fork();if(pid > 0){// 父进程逻辑close(pipefd[0]); // 关闭读端rubbish.push_back(pipefd[1]);// 保存管道写端、PID和进程名channels.push_back(channel(pipefd[1], pid, "Process-" + to_string(i)));}else if(pid == 0){// 子进程逻辑：关闭继承的多余管道cout << "rubbish process: ";for(const auto& e : rubbish){cout << e << " ";close(e); }cout << endl;close(pipefd[1]); // 关闭写端dup2(pipefd[0], 0); // 重定向标准输入close(pipefd[0]); slaver(tasks); // 执行任务循环exit(0); // 子进程退出}else{perror("fork error");}}
}

五、任务分配函数

void ctrlslaver(const vector<channel>& channels, const vector<function<void()>>& tasks)
{int cmdcode = 0; // 任务编号for(int i = 0; i < TASKNUM; i++){// 随机选择一个子进程int process_index = rand() % channels.size();// 发送任务信息cout << "parent says: send the message " << cmdcode << " to process: " << channels[process_index].getSlaverId()<< " 第" << i << "次发送" << endl;sleep(2);// 向选中进程发送任务编号write(channels[process_index].getCwdFd(), &cmdcode, sizeof(int));// 循环使用任务列表cmdcode = (cmdcode + 1) % tasks.size(); }
}

六、进程池销毁机制

请注意，这里是重点，实际上由于子进程不断创建，进程池的结构是这样的：

在创建完成第一个子进程时，父进程留有一个写端口，如果继续创建子进程，那么写端口会被保留下来，后果就是每一个子进程都会有指向某个管道的写端口，因此需要销毁机制！

// 反向回收法：解决管道关闭阻塞问题
void Method1(const vector<channel>& channels)
{auto rit = channels.rbegin();while(rit != channels.rend()){cout << "Closing process: " << rit->getSlaverId() << endl;close(rit->getCwdFd()); // 关闭写端waitpid(rit->getSlaverId(), nullptr, 0); // 等待子进程退出rit++;}
}// 进程池销毁入口
void destroyprocesspool(const vector<channel>& channels)
{Method1(channels); // 使用反向回收法
}

七、主函数入口

int main()
{srand((unsigned int)time(nullptr));vector<channel> channels; // 管道通道列表vector<function<void()>> tasks; // 任务队列initprocesspool(channels, tasks); // 初始化进程池sleep(2);ctrlslaver(channels, tasks); // 分配任务destroyprocesspool(channels); // 销毁进程池cout << "Process pool destroyed." << endl;return 0;
}

命名管道：

命名管道的原理：

匿名管道应⽤的⼀个限制就是只能在具有共同祖先（具有亲缘关系）的进程间通信。
如果我们想在不相关的进程之间交换数据，可以使⽤FIFO⽂件来做这项⼯作，它经常被称为命名管道。
命名管道是⼀种特殊类型的⽂件。
上面的讲的是匿名管道，即没有文件名，系统通过pipe接口分配一块内存给具有血缘关系的进程进行通信，如果是两个互不相关的进程进行通信，就需要创建命名管道
进程间通信的本质是需要看到同一块资源，如果两个进程读写同一个文件，那么架构上与匿名管道大致相同，命名管道如何确保是打开的同一个文件？路径+文件名一致确保

命名管道的创建：

可以使用命令行创建或者函数接口创建：

mkfifo filename

pathname是创建mkfifo文件的路径，mode是文件的权限；

如果想要删除mkfifo文件，可以使用unlink:

实战运用：

下面是一个关于unlink和mkfifo函数运用的小测试：

#include <iostream>
#include <unistd.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <fcntl.h>
#include <wait.h>
#include <cstring>
#include <cstdlib>using namespace std;// 定义管道名称
const char* FIFO_NAME = "/tmp/my_test_fifo";// 写入数据到管道
void write_to_fifo() {// 打开管道（写模式）int fd = open(FIFO_NAME, O_WRONLY);if (fd == -1) {perror("open fifo for writing failed");exit(EXIT_FAILURE);}// 要写入的数据const char* message = "Hello from write process!";int bytes_written = write(fd, message, strlen(message) + 1);if (bytes_written == -1) {perror("write to fifo failed");close(fd);exit(EXIT_FAILURE);}cout << "Writer: Successfully wrote " << bytes_written << " bytes" << endl;close(fd);
}// 从管道读取数据
void read_from_fifo() {// 打开管道（读模式）int fd = open(FIFO_NAME, O_RDONLY);if (fd == -1) {perror("open fifo for reading failed");exit(EXIT_FAILURE);}// 读取数据char buffer[1024];int bytes_read = read(fd, buffer, sizeof(buffer));if (bytes_read == -1) {perror("read from fifo failed");close(fd);exit(EXIT_FAILURE);}cout << "Reader: Received message: " << buffer << endl;close(fd);
}int main() {// 1. 使用mkfifo创建命名管道mode_t mode = 0666; // 管道权限if (mkfifo(FIFO_NAME, mode) == -1) {perror("mkfifo failed");// 如果管道已存在，也可以继续执行if (errno != EEXIST) {exit(EXIT_FAILURE);}cout << "FIFO already exists, proceeding..." << endl;} else {cout << "Successfully created FIFO: " << FIFO_NAME << endl;}// 2. 创建子进程进行通信pid_t pid = fork();if (pid == -1) {perror("fork failed");unlink(FIFO_NAME); // 清理管道文件exit(EXIT_FAILURE);}if (pid == 0) {// 子进程：读取数据read_from_fifo();exit(EXIT_SUCCESS);} else {// 父进程：写入数据sleep(1); // 确保子进程先打开管道write_to_fifo();wait(nullptr); // 等待子进程结束// 3. 使用unlink删除管道文件if (unlink(FIFO_NAME) == -1) {perror("unlink failed");exit(EXIT_FAILURE);}cout << "Successfully removed FIFO: " << FIFO_NAME << endl;}return 0;
}