【Linux】进程通信之管道

修修修也

发布于 2024-11-17 12:26:36

31000

代码可运行

文章被收录于专栏：修也的进阶日记修也的进阶日记

运行总次数：0

代码可运行

📌什么是管道(pipe)

管道是Unix中最古老的进程间通信的形式。它是进程之间的一个单向数据流 : 一个进程写入管道的所有数据都由内核定向到另一个进程,另一个进程由此就可以从管道中读取数据。管道示意图如下:

其实管道的概念非常好理解, 举个例子, 相信大家小时候身边都或多或少有过订早餐奶的经历, 送奶员工每天早上定时将当天新鲜的鲜奶放进家门口的奶箱, 然后大家睡醒就可以在奶箱里拿到新鲜的牛奶。其中奶箱就相当于"管道", 送奶工放入牛奶就相当于给管道里写入数据,而客户取出牛奶就相当于读取数据。

📌在Linux中使用管道

首先,学过Linux命令的话,大家对于管道肯定不陌生, Linux管道使用竖线 | 连接多个命令,这个被称为管道符。如:

$ ls | head -5

以上这行代码就组成了一个管道，它的功能是将前一个命令（ls）的输出，作为后一个命令（head -5）的输入:(原本ls指令会打印当前目录所有文件/目录名,但是通过管道和head命令后就变为只打印前5行文件/目录名了)

从这个功能描述中，我们可以看出管道中的数据只能单向流动，也就是半双工通信，如果想实现相互通信（全双工通信），我们需要创建两个管道才行。

另外，通过管道符 | 创建的管道是匿名管道，用完了就会被自动销毁。需要注意的是，匿名管道只能在具有亲缘关系（父子进程,兄弟进程,爷孙进程）的进程间使用。也就是说，匿名管道只能用于亲缘进程之间的通信。

📌管道的原理

要构建一个父子进程共享的单项管道文件,我们就先在父进程中以读和写两种方式打开同一个管道文件:

然后我们再创建子进程, 这时子进程就会拷贝父进程的文件控制结构体以及里面的文件指针数据,即它们会指向相同的文件:

然后我们分别关闭父进程对管道文件的读方式,以及子进程对管道文件的写方式,这时,管道文件就成为了一个由父进程写入数据,子进程读取数据的通信管道文件:

当然也可以关闭父进程对管道文件的写方式,以及子进程对管道文件的读方式,这样管道文件就成为了一个由子进程写入数据,父进程读取数据的通信管道文件。

📌创建管道

🎏创建管道接口

因为管道是内存级文件,并非磁盘级文件,所以当我们想创建一个管道时,不能使用open()函数来打开文件,而是要使用pipe()函数,下面是pipe()函数的手册:

函数定义:

 int pipe(int pipefd[2]);

函数参数:

int pipefd[2]

这个参数是一个输出型参数,作用是把我们分别以读方式和写方式打开的文件的文件描述符数字带出来让用户使用。其中, pipefd[0]为读下标, pipefd[1]为写下标。

函数返回值:

int

当函数打开管道文件成功后, 返回0; 出错时, 则返回-1。

🎏编码实现父子进程通信管道

管道的实现思路如下:

创建管道
创建子进程, 子进程关闭读端, 然后开始向管道写入数据
父进程关闭写端,然后开始向管道读数据
读取完毕,父子进程关闭自己所使用的写/读端

综上,实现代码如下:

#include<iostream>
#include<cstdio>
#include<cstring>
#include<cstdlib>
#include<unistd.h>
#include<sys/types.h>
#include<sys/wait.h>

#define N 2
#define NUM 1024

using namespace std;

//child
void Writer(int wfd)
{
    string s="hello,I am child";
    pid_t self = getpid();
    int number = 0;

    char buffer[NUM];
    while(true)
    {
        //构建发送字符串
        buffer[0] = 0;  //字符串清空,仅作提醒,把这个数组当字符串

        //字符串安全格式接口
        snprintf(buffer,sizeof(buffer),"%s-%d-%d",s.c_str(),self,number++);
        //把后面的几个参数按照引号内的格式拼在第一个参数的数组里,长度为第二个参数

        //cout << buffer << endl;

        //发送/写入给父进程
        write(wfd,buffer,strlen(buffer));

        sleep(1);
    }
}

//father
void Reader(int rfd)
{
    char buffer[NUM];
    while(true)
    {
        buffer[0] = 0;
        ssize_t n = read(rfd,buffer,sizeof(buffer));
        if(n > 0)
        {
            buffer[n] = 0;//0 =='\0'
            cout<<"father get a message["<<getpid()<<"]#"<<buffer<<endl;
        }
    }
}

int main()
{
    //创建管道
    int pipefd[N] = {0};
    int n = pipe(pipefd);
    if(n < 0)   return 1;

    // cout<<pipefd[0]<<":"<<pipefd[1]<<endl;

    //创建子进程
    //建立单向信道   child写father读
    pid_t id = fork();
    if(id < 0) return 2;
    if(id == 0)
    {
        //child写关闭读
        close(pipefd[0]);

        //IPC code
        Writer(pipefd[1]);

        //用完把写端也关掉
        close(pipefd[1]);
        exit(0);
    }

    //father读关闭写
     close(pipefd[1]);

    //IPC code
    Reader(pipefd[0]);

    pid_t rid = waitpid(id,nullptr,0);
    if(rid < 0) return 3;

    //用完把读端也关掉
     close(pipefd[0]);

    return 0;
}

运行程序,可以发现父进程可以接收到子进程传来的消息了:

📌管道读写规则

🎏当读写端正常但读端没有数据可读时

O_NONBLOCK disable：read调用阻塞，即进程暂停执行，一直等到有数据来到为止。
O_NONBLOCK enable：read调用返回-1，errno值为EAGAIN。

借助我们上面的程序进行验证: 我们让子进程每隔1秒进行写入,而父进程一直在做读操作,运行程序,发现父进程没有疯狂打印消息,而是也每隔一秒打印一次,这说明当管道中没有数据可读时,读取端(read)是会调用阻塞,暂停进程执行,直到有数据来为止:

🎏当读写端正常但写入时管道还是满的时

O_NONBLOCK disable： write调用阻塞，直到有进程读走数据
O_NONBLOCK enable：调用返回-1，errno值为EAGAIN

借助程序进行验证:

我们让子进程一直进行写入, 而父进程每隔5秒做读操作, 运行程序, 发现父进程每隔5秒会打印一大批消息, 而且这些消息是连续的, 并没有断层, 这说明当管道中数据满了的时候, 写入端(write)是会调用阻塞,暂停进程执行,直到所有数据被读取走后才会继续写入:

验证代码如下:

#include<iostream>
#include<cstdio>
#include<cstring>
#include<cstdlib>
#include<unistd.h>
#include<sys/types.h>
#include<sys/wait.h>

#define N 2
#define NUM 1024

using namespace std;

//child
void Writer(int wfd)
{
    string s="hello,I am child";
    pid_t self = getpid();
    int number = 0;

    char buffer[NUM];
    while(true)
    {
        //构建发送字符串
        buffer[0] = 0;  //字符串清空,仅作提醒,把这个数组当字符串

        //字符串安全格式接口
        snprintf(buffer,sizeof(buffer),"%s-%d-%d",s.c_str(),self,number++);
        //把后面的几个参数按照引号内的格式拼在第一个参数的数组里,长度为第二个参数

        //cout << buffer << endl;

        //发送/写入给父进程
        write(wfd,buffer,strlen(buffer));

        //sleep(1);

    }

}

//father
void Reader(int rfd)
{
    char buffer[NUM];
    while(true)
    {
        buffer[0] = 0;
        ssize_t n = read(rfd,buffer,sizeof(buffer));
        if(n > 0)
        {
            buffer[n] = 0;//0 =='\0'
            cout<<"father get a message["<<getpid()<<"]#"<<buffer<<endl;
        }

        sleep(5);
    }
}


int main()
{
    //创建管道
    int pipefd[N] = {0};
    int n = pipe(pipefd);
    if(n < 0)   return 1;

    // cout<<pipefd[0]<<":"<<pipefd[1]<<endl;

    //创建子进程
    //建立单向信道   child写father读
    pid_t id = fork();
    if(id < 0) return 2;
    if(id == 0)
    {
        //child写关闭读
        close(pipefd[0]);

        //IPC code
        Writer(pipefd[1]);


        //用完把写端也关掉
        close(pipefd[1]);
        exit(0);
    }

    //father读关闭写
     close(pipefd[1]);

    //IPC code
    Reader(pipefd[0]);

    pid_t rid = waitpid(id,nullptr,0);
    if(rid < 0) return 3;


    //用完把读端也关掉
     close(pipefd[0]);

    return 0;
}

🎏当管道写端对应的文件描述符被关闭,读端正常时

如果所有管道写端对应的文件描述符被关闭，则read返回0

我们验证一下,让子进程每一秒写一个字符,写5个后就退出,看看运行结果:

结果是当子进程写端被关闭后,父进程的read()就会一直疯狂打印read()的返回值0,表明读到了文件(pipe)结尾,不会被阻塞.

验证代码如下:

#include<iostream>
#include<cstdio>
#include<cstring>
#include<cstdlib>
#include<unistd.h>
#include<sys/types.h>
#include<sys/wait.h>

#define N 2
#define NUM 1024

using namespace std;

//child
void Writer(int wfd)
{
    string s="hello,I am child";
    pid_t self = getpid();
    int number = 0;

    char buffer[NUM];
    while(true)
    {
        sleep(1);
        //构建发送字符串
        //buffer[0] = 0;  //字符串清空,仅作提醒,把这个数组当字符串

        //字符串安全格式接口
        //snprintf(buffer,sizeof(buffer),"%s-%d-%d",s.c_str(),self,number++);
        //把后面的几个参数按照引号内的格式拼在第一个参数的数组里,长度为第二个参数

        //cout << buffer << endl;

        //发送/写入给父进程
        //write(wfd,buffer,strlen(buffer));

        //每次只写一字节,写完后计数
        char c = 'c';
        write(wfd,&c,1);
        number++;
        cout<<number<<endl;
        //sleep(1);

        //写端写5个字符就退出
        if(number >= 5) break;

    }

}

//father
void Reader(int rfd)
{
    char buffer[NUM];
    while(true)
    {
        
        buffer[0] = 0;
        ssize_t n = read(rfd,buffer,sizeof(buffer));
        if(n > 0)
        {
            buffer[n] = 0;//0 =='\0'
            cout<<"father get a message["<<getpid()<<"]#"<<buffer<<endl;
        }
        // else if(n == 0) 
        // {
        //     cout<<"father read file done!\n"<<endl;
        //     break;  //说明读到了文件末尾,可以返回了        
        // }
        // else break;

        // sleep(5);
        //cout<<"n:"<<n<<endl;
    }
}


int main()
{
    //创建管道
    int pipefd[N] = {0};
    int n = pipe(pipefd);
    if(n < 0)   return 1;

    // cout<<pipefd[0]<<":"<<pipefd[1]<<endl;

    //创建子进程
    //建立单向信道   child写father读
    pid_t id = fork();
    if(id < 0) return 2;
    if(id == 0)
    {
        //child写关闭读
        close(pipefd[0]);

        //IPC code
        Writer(pipefd[1]);


        //用完把写端也关掉
        close(pipefd[1]);
        exit(0);
    }

    //father读关闭写
     close(pipefd[1]);

    //IPC code
    Reader(pipefd[0]);

    pid_t rid = waitpid(id,nullptr,0);
    if(rid < 0) return 3;


    //用完把读端也关掉
     close(pipefd[0]);

    return 0;
}

🎏当管道读端对应的文件描述符被关闭,写端正常时

如果所有管道读端对应的文件描述符被关闭，则write操作会产生信号SIGPIPE,进而可能导致write进程退出

我们借助程序验证一下:

验证代码如下:

#include<iostream>
#include<cstdio>
#include<cstring>
#include<cstdlib>
#include<unistd.h>
#include<sys/types.h>
#include<sys/wait.h>

#define N 2
#define NUM 1024

using namespace std;

//child
void Writer(int wfd)
{
    string s="hello,I am child";
    pid_t self = getpid();
    int number = 0;

    char buffer[NUM];
    while(true)
    {
        sleep(1);
        //构建发送字符串
        buffer[0] = 0;  //字符串清空,仅作提醒,把这个数组当字符串

        //字符串安全格式接口
        snprintf(buffer,sizeof(buffer),"%s-%d-%d",s.c_str(),self,number++);
        //把后面的几个参数按照引号内的格式拼在第一个参数的数组里,长度为第二个参数

        //cout << buffer << endl;

        //发送/写入给父进程
        write(wfd,buffer,strlen(buffer));

        //每次只写一字节,写完后计数
        // char c = 'c';
        // write(wfd,&c,1);
        // number++;
        // cout<<number<<endl;
        sleep(1);

        //写端写5个字符就退出
        //if(number >= 5) break;

    }

}

//father
void Reader(int rfd)
{
    char buffer[NUM];
    int cnt = 0;
    while(true)
    {
        
        buffer[0] = 0;
        ssize_t n = read(rfd,buffer,sizeof(buffer));
        if(n > 0)
        {
            buffer[n] = 0;//0 =='\0'
            cout<<"father get a message["<<getpid()<<"]#"<<buffer<<endl;
        }
        else if(n == 0) 
        {
            cout<<"father read file done!\n"<<endl;
            break;  //说明读到了文件末尾,可以返回了        
        }
        else break;

        // sleep(5);
        //cout<<"n:"<<n<<endl;
        cnt++;
        if(cnt > 5)
        {
            break;
        }
    }
}


int main()
{
    //创建管道
    int pipefd[N] = {0};
    int n = pipe(pipefd);
    if(n < 0)   return 1;

    // cout<<pipefd[0]<<":"<<pipefd[1]<<endl;

    //创建子进程
    //建立单向信道   child写father读
    pid_t id = fork();
    if(id < 0) return 2;
    if(id == 0)
    {
        //child写关闭读
        close(pipefd[0]);

        //IPC code
        Writer(pipefd[1]);


        //用完把写端也关掉
        close(pipefd[1]);
        exit(0);
    }

    //father读关闭写
     close(pipefd[1]);

    //IPC code
    Reader(pipefd[0]);  //读取5s
    //用完把读端也关掉
     close(pipefd[0]);
     cout<<"father close read fd:"<<pipefd[0]<<endl;

    sleep(3);

    int status = 0;
    pid_t rid = waitpid(id,&status,0);
    if(rid < 0) return 3;
    cout<<"wait child success:"<<rid<<"exit code:"<< ((status>>8)&0xFF) << "exit signal:" << (status&0x7F) <<endl;


    //用完把读端也关掉
     close(pipefd[0]); 

    return 0;
}

tips:

当要写入的数据量不大于PIPE_BUF时，linux将保证写入的原子性
当要写入的数据量大于PIPE_BUF时，linux将不再保证写入的原子性

📌管道特点

管道的特点:

只能用于具有共同祖先的进程（具有亲缘关系的进程）之间进行通信；通常，一个管道由一个进程创建，然后该进程调用fork，此后父、子进程之间就可应用该管道。
管道提供流式服务
管道是基于文件的, 而文件的生命周期是随进程的, 一般而言，进程退出，管道释放，所以管道的生命周期随进程
一般而言，内核会对管道操作进行同步与互斥, 以此来保护管道文件的数据安全
管道是半双工的，数据只能向一个方向流动；需要双方通信时，需要建立起两个管道

tips:

管道是有固定大小的,在不同内核里,大小可能有差别,我们调用ulimit指令查询得到的结果如下:

查询pipe的手册,我们可以知道在不同的内核版本下,管道的大小也不相同:

再使用程序验证时,我们会发现子进程会给父进程写入65536个字节的数据,即64KB数据,这说明我们当前环境下管道的大小是64KB:

验证代码如下:

#include<iostream>
#include<cstdio>
#include<cstring>
#include<cstdlib>
#include<unistd.h>
#include<sys/types.h>
#include<sys/wait.h>

#define N 2
#define NUM 1024

using namespace std;

//child
void Writer(int wfd)
{
    string s="hello,I am child";
    pid_t self = getpid();
    int number = 0;

    char buffer[NUM];
    while(true)
    {
        //构建发送字符串
        //buffer[0] = 0;  //字符串清空,仅作提醒,把这个数组当字符串

        //字符串安全格式接口
        //snprintf(buffer,sizeof(buffer),"%s-%d-%d",s.c_str(),self,number++);
        //把后面的几个参数按照引号内的格式拼在第一个参数的数组里,长度为第二个参数

        //cout << buffer << endl;

        //发送/写入给父进程
        //write(wfd,buffer,strlen(buffer));

        //每次只写一字节,写完后计数
        char c = 'c';
        write(wfd,&c,1);
        number++;
        cout<<number<<endl;
        //sleep(1);
    }

}

//father
void Reader(int rfd)
{
    char buffer[NUM];
    while(true)
    {
        sleep(50);
        buffer[0] = 0;
        ssize_t n = read(rfd,buffer,sizeof(buffer));
        if(n > 0)
        {
            buffer[n] = 0;//0 =='\0'
            cout<<"father get a message["<<getpid()<<"]#"<<buffer<<endl;
        }

        // sleep(5);
    }
}


int main()
{
    //创建管道
    int pipefd[N] = {0};
    int n = pipe(pipefd);
    if(n < 0)   return 1;

    // cout<<pipefd[0]<<":"<<pipefd[1]<<endl;

    //创建子进程
    //建立单向信道   child写father读
    pid_t id = fork();
    if(id < 0) return 2;
    if(id == 0)
    {
        //child写关闭读
        close(pipefd[0]);

        //IPC code
        Writer(pipefd[1]);


        //用完把写端也关掉
        close(pipefd[1]);
        exit(0);
    }

    //father读关闭写
     close(pipefd[1]);

    //IPC code
    Reader(pipefd[0]);

    pid_t rid = waitpid(id,nullptr,0);
    if(rid < 0) return 3;


    //用完把读端也关掉
     close(pipefd[0]);

    return 0;
}