管道和消息队列

提示：文章写完后，目录可以自动生成，如何生成可参考右边的帮助文档

一、管道（Pipe）

管道是一种Linux中的进程间通信的方式（IPC），通过在两个进程之间创建一个单向数据通道，实现数据传输。

管道的分类

无名管道

特点：

• 只能在有亲缘关系的进程间使用：父子进程或者兄弟进程。
• 半双工通信：数据只能单向流动，不能同时向两端输出。
• 基于文件描述符，通过pipe（）函数创建。
• 存在于内存中，随进程结束而销毁。

1.创建方式

intret;intfd[2];//用数组存储数据，fd[0]为读端，fd[1]为写端。ret=pipe(fd);//创建匿名管道

2.使用方法

在父进程中通过fork（）创建子进程，分别在父子进程中实现不同的功能，同时在使用时需要关闭不需要的进程。例如为写端时，需要close（fd[0])关闭读端。

3.使用管道实现 ps aux | grep bash 指令

//create pipeintfd[2];intret=pipe(fd);if(ret==-1){printf("failed!\n");exit(1);}pid_tpid;pid=fork();//father ps auxif(pid>0){close(fd[0]);dup2(fd[1],STDOUT_FILENO);execlp("ps","ps","aux",NULL);//replace father processperror("execlp");exit(0);}//child grep bashelseif(pid==0){close(fd[1]);dup2(fd[0],STDIN_FILENO);execlp("grep","grep","bash",NULL);//replace child processperror("execlp");exit(0);}// when fork failed,this command will be excutedprintf("fd[0] is %d,\n",fd[0]);printf("fd[1] is %d,\n",fd[1]);close(fd[0]);close(fd[1]);return0;

首先先用一个数组创建管道的读端和写端，再用这个数组创建一个匿名管道。再通过fork()函数创建子进程，通过execlp函数分别对父子进程分配工作，父进程完成ps aux命令，通过dup2重定向写入到管道中。而子进程通过重定向从管道中读取数据，在执行grep bash，筛选出grep找到包含"bash"的行，写入自己的标准输出（默认是屏幕）。

命名管道

特点：

• 允许无亲缘关系的进程间通信
• 命名管道以文件形式存在于文件系统中
• 半双工通信

1.命名管道的创建

#include<sys/types.h>#include<sys/stat.h>intmkfifo(constchar*pathname,mode_tmode);

mkfifo函数需要两个参数，一个是管道名，一个是文件权限。创建完成后会在文件中显示一个新类型的文件。

中间mkfifo即为管道类型的文件，文件类型是p开头。

intret;ret=mkfifo("./mymkfifo",0777);if(ret==-1){printf("mkfifo failed!\n");return-1;}printf("mkfifo success!\n");

上述代码即可创建一个管道文件。

2.命名管道的使用

要想使用管道，需要给管道创建文件描述符。
实现read端：

intfd;intaread;charreadbuf[50]={0};fd=open("./mymkfifo",O_RDONLY);//以只读权限打开管道if(fd==-1){//打开失败，管道不存在printf("open failed!\n");return-1;}aread=read(fd,readbuf,50);//read函数返回读到的字节数printf("read %d byte from fifo:%s.",aread,readbuf);close(fd);return0;

实现write端

intfd;charstr[128]={0};//存储写的内容fd=open("./mymkfifo",O_WRONLY);//以只写权限打开if(fd==-1){printf("open failed!\n");return-1;}fgets(str,sizeof(str),stdin);//put from keyboardwrite(fd,str,strlen(str));//写入close(fd);return0;

这样即可实现数据的单向流动。
运行一下试试：

二、消息队列（Message Queue）

消息队列是消息的链表，允许多个进程向队列中写入消息或读取消息。

1.消息队列函数解析

ftok() - 键值生成

key_tkey;// 键值key可以用来控制两个进程在同一个消息队列中进行数据操作。/* 常用方式：使用当前目录和项目ID */key=ftok(".",'A');// 使用当前目录和字符'A'生成键值/* 注意事项 */// 1. path必须存在且可访问// 2. id只使用低8位（0-255）// 3. 删除文件后重建会得到不同的key

msgget() - 队列创建/获取

intmsqid;/* 创建私有队列（用于父子进程） */msqid=msgget(IPC_PRIVATE,IPC_CREAT|0666);/* 创建或获取命名队列 */key=ftok("/tmp",'M');msqid=msgget(key,IPC_CREAT|IPC_EXCL|0666);// 排他性创建/* msgflg标志详解 */#defineIPC_CREAT01000/* 不存在则创建 */#defineIPC_EXCL02000/* 与IPC_CREAT配合，排他创建 */#defineIPC_NOWAIT04000/* 非阻塞操作（用于后续调用） *//* 权限位：类似文件权限 */// 0600 - 所有者读写// 0666 - 所有用户读写

msgsnd() - 发送消息

structmy_msgbuf{longmtype;chardata[100];}msg;msg.mtype=1;// 类型必须 > 0strcpy(msg.data,"Hello World");/* 阻塞发送 */if(msgsnd(msqid,&msg,sizeof(msg.data),0)==-1){perror("msgsnd");}/* 非阻塞发送 */if(msgsnd(msqid,&msg,sizeof(msg.data),IPC_NOWAIT)==-1){if(errno==EAGAIN){printf("队列满，稍后重试\n");}}/* msgflg标志 */#defineIPC_NOWAIT04000/* 非阻塞操作 */

msgrcv() - 接收消息（核心重点）

structmy_msgbufmsg;ssize_trecv_len;/* msgtyp参数详解 - 消息接收规则 */recv_len=msgrcv(msqid,&msg,sizeof(msg.data),msgtyp,msgflg);// msgtyp = 0 : 接收队列中第一个消息（FIFO）// msgtyp > 0 : 接收类型等于msgtyp的第一个消息// msgtyp < 0 : 接收小于等于|msgtyp|的最小类型值（优先级模型）/* 实际应用场景 */// 场景1：按顺序接收所有消息msgrcv(msqid,&msg,size,0,0);// 最简单的FIFO// 场景2：只接收特定类型的消息（如只处理错误日志）msgrcv(msqid,&msg,size,2,0);// 只接收类型为2的消息// 场景3：实现优先级处理（数值越小优先级越高）msgrcv(msqid,&msg,size,-10,0);// 接收类型1-10中最小类型的消息// 场景4：接收任何消息，但不阻塞if(msgrcv(msqid,&msg,size,0,IPC_NOWAIT)==-1){if(errno==ENOMSG){printf("队列为空\n");}}/* msgflg标志 */#defineIPC_NOWAIT04000/* 非阻塞 */#defineMSG_NOERROR010000/* 消息过长时截断而非报错 */

msgctl() - 控制操作

structmsqid_dsbuf;/* cmd命令详解 */// 1. IPC_STAT - 获取队列状态if(msgctl(msqid,IPC_STAT,&buf)==0){printf("队列中消息数: %ld\n",buf.msg_qnum);printf("队列最大字节: %ld\n",buf.msg_qbytes);printf("最后发送进程: %d\n",buf.msg_lspid);}// 2. IPC_SET - 修改队列属性buf.msg_perm.mode=0666;// 修改权限buf.msg_qbytes=65536;// 修改队列大小（需要特权）if(msgctl(msqid,IPC_SET,&buf)==-1){perror("msgctl IPC_SET");}// 3. IPC_RMID - 删除消息队列if(msgctl(msqid,IPC_RMID,NULL)==-1){perror("msgctl IPC_RMID");}