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

系统编程——消息队列

news 2025/8/12 6:32:26

前言

进程间通信（Inter process communication，简称IPC）指的是进程之间的信息交换，进程间通信的方式有很多，比如管道通信、信号通信、共享内存、消息队列、信号量组、POSIX信号量等。
在这里插入图片描述

一、消息队列概念

Linux系统中消息队列（Message Queue）是进程间通信的一种方式，这种通信机制的好处是可以传输指定类型(用户可以自行定义)的数据，相同类型的数据根据到达顺序在队列中进行排队。

当然，不同类型的数据不能处于同一个队列中，也就是说系统中可能存在多个消息队列，每个消息队列中的数据类型都是不同的，所以用户打算读取消息队列中的数据时也需要指定数据类型，才可以从存储该类型数据的消息队列中读取有效数据。
在这里插入图片描述
Linux系统每个创建的消息队列都具有一个唯一的键值key，进程可以通过指定消息队列的键值来向消息队列发送数据。Linux系统中提供了一个shell命令：ipcs -a来查看系统中所有的IPC对象的信息。

二、创建消息队列

Linux系统提供了一个名称叫做msgget()的函数接口，用户利用该接口可以创建或者打开一个消息队列，同样Linux系统中提供了一个shell命令：ipcmk 可以用于创建IPC对象。
(1)ipcmk命令
在这里插入图片描述

(2)msgget()函数

msgget函数有两个参数，第一个参数需要传入一个key_t类型的值，该值指的是要创建的消息队列的key键值，key也可以称为密钥。键值类型key_t其实在内核源码中指的是int类型，如下图：

msgget()函数的第二个参数指的是创建消息队列的标志，其中IPC_CREAT指的是如果消息队列不存在则创建，IPC_EXCL指的是如果消息队列存在则表示函数调用失败。

另外，也可以指定消息队列的权限，权限的结构和open函数的mode类型，采用八进制表示，只不过消息队列不需要设置执行权限，所以权限设置为0644即可。

Linux系统中提供了一个名称叫做ftok()的函数接口，利用该接口可以生成IPC对象的键值key，该接口的使用规则如下：
在这里插入图片描述
可以看到，ftok()函数可以把一个指定路径的文件和一个指定的项目id转换为一个system-V IPC对象使用的键值key。

在man手册中提到ftok()函数生成的键值key的组成：proj_id的低8位+ 设备编号的低8位+ inode编号的低16位。

三、访问消息队列

Linux系统中提供了一个名称叫做msgsnd()的函数接口，用户利用该函数可以向指定的消息队列发送消息，同时提供了一个名称叫做msgrcv()的函数接口，用户利用该接口可以从指定的消息队列中读取消息。

(1)发送消息

在这里插入图片描述

msgsnd函数的第一个参数msgid指的是消息队列的标识符，该标识符可以通过msgget函数得到。

msgsnd函数的第二个参数msgp指的是一个指向struct msgbuf类型的结构体指针，该结构体中有两个成员，其中一个成员mtype指的是消息类型，必须是一个大于0的正整数，另一个成员mtext指的是消息正文，类型可以是数组或者其他结构。
在这里插入图片描述

msgsnd函数的第三个参数msgsz指的是消息正文的大小，按字节计算，当然msgsz的值必须是非负整数，可以设置为0，表示消息正文的长度为0。

msgsnd函数的第四个参数msgflg指的是消息队列的标志，如果该标志设置为IPC_NOWAIT，则表示不阻塞，此时如果待写入的消息的长度大于消息队列剩余空间，则直接返回并报错。

注意：消息队列默认的属性是阻塞的，也就是当待写入的消息的长度大于消息队列剩余空间时，默认阻塞，直到消息队列的容量足够容纳时会解除阻塞。

(2)读取消息

在这里插入图片描述
第一个参数：msgqid指的是MSG对象的标识符ID，MSG标识符可以通过msgget()函数获取。

第二个参数：msgp指的是存放消息的缓存地址，该地址下存储的是struct msgbuf结构体。

第三个参数：msgsz指的是存放消息的缓存的大小，按照字节计算，如果消息正文的大小大于用户设置的缓存大小，则根据msgflg是否为MSG_NOERROR进行判断，如果msgflg设置为MSG_NOERROR ，则可以读取对应字节的消息，如果msgflg没有设置，则无法读取消息并报错。

第四个参数：msgtyp指的是要接收消息的类型，在调用msgsnd函数时构造的消息结构体中有该成员的值。

1)等于0：指的是不区分类型，直接读取MSG中的第一个消息。

2)大于0：读取类型为指定msgtyp的第一个消息（若msgflg被配置了MSG_EXCEPT则读取除了类型为msgtyp的第一个消息）。

3)小于0：读取类型小于等于msgtyp绝对值的第一个具有最小类型的消息。例如当MSG对象中有类型为3、1、5类型消息若干条，当msgtyp为-3时，类型为3的第一个消息将被读取。

第五个参数：msgflg指的是接收消息选项，如果msgflg设置为0，指的是默认接收模式，在MSG中无指定类型消息时阻塞。

1)IPC_NOWAIT ：指的是非阻塞接收模式，当MSG中没有指定类型消息时直接退出函数
2)MSG_EXCEPT ：指的是读取除msgtyp之外的第一个消息。
3)MSG_NOERROR：如果待读取的消息尺寸比msgsz大只返回msgsz部分，其余部分丢弃

四、控制消息队列

IPC对象是一种持久性资源，如果没有明确的删除掉IPC对象，则IPC对象是不会自动从内存中消失的。用户除了可以使用命令的方式删除，也可以使用函数来删除。

Linux系统中提供了一个名称叫做msgctl的函数接口，用户可以利用该函数实现获取消息队列的属性信息、设置消息队列的属性信息、删除消息队列等操作。

1.指令删除
在这里插入图片描述
2.函数删除

五、练习

1.

设计程序，在Linux系统中创建一个消息队列，并测试消息队列的键值key的组成是否正确。提示：可以通过stat()函数获取文件的属性信息。

/** Copyright (c)* * date: 2025-7-30* * author: Charles* * function name : msg_que.c** function: 设计程序，在Linux系统中创建一个消息队列，并测试消息队列的键值key的组成是* 否正确。提示：可以通过stat()函数获取文件的属性信息。**/#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <stdio.h>
#include <errno.h>
#include <string.h>int main(int argc, char const *argv[])
{	key_t key = ftok(".", 0xFFFFFF01);int msg_id = msgget(key, IPC_CREAT);if(-1 == msg_id){fprintf(stderr, "msgget error,errno:%d,%s\n", errno,strerror(errno));}printf("msg key = %#010x\n", key);struct stat file_stat;stat(".", &file_stat);printf("%#010lx\n", file_stat.st_dev);printf("%#010lx\n", file_stat.st_ino);return 0;
}

2.

要求进程A创建一条消息队列之后向进程B发送SIGUSR1信号，进程B收到该信号之后打开消息队列并把进程的PID作为消息写入到消息队列中，要求进程B在写入消息之后，发SIGUSR2信号给进程A，
进程A收到该信号则从消息队列中读取消息并输出消息正文的内容。

/** Copyright (c)* * date: 2025-7-30* * author: Charles* * function name : msg_sig.c** function: 要求进程A创建一条消息队列之后向进程B发送SIGUSR1信号，进程B收到该信号之后打开消息队列并把进程的PID作为消息写入到消息队列中，* 要求进程B在写入消息之后，发SIGUSR2信号给进程A，进程A收到该信号则从消息队列中读取消息并输出消息正文的内容。  **/#include <unistd.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <sys/wait.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <time.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <signal.h>key_t key;
int msg_id;struct msg_buf
{long msg_type; //消息类型，必须大于0int msg_text;	//消息文本
};void signal_handler(int signo)
{if(signo == SIGUSR1){struct msg_buf send_buf;send_buf.msg_type = 1;send_buf.msg_text = getpid();msgsnd(msg_id, &send_buf, 4, 0);//默认不阻塞printf("Sent message to parent: %d\n", send_buf.msg_text);kill(getppid(), SIGUSR2);	//为什么在这里发消息，确保子进程已经写完信息在发信号给父进程}else if(signo == SIGUSR2){struct msg_buf recv_buf;msgrcv(msg_id, &recv_buf, 4, 0, 0);printf("Received message from child: %d\n", recv_buf.msg_text);}
}int main(int argc, char const *argv[])
{//获取消息队列的键key = ftok(".", 0xFFFFFF01);//获取消息队列的ID msg_id = msgget(key, IPC_CREAT | 0644);if(-1 == msg_id){fprintf(stderr, "msgget error,errno:%d,%s\n", errno, strerror(errno));}pid_t pid = fork();if(pid > 0){signal(SIGUSR2, signal_handler);sleep(1);		//一定要等到子进程写入消息，再发信号，避免丢失消息kill(pid, SIGUSR1);pause();		//没有捕获到信号就一直阻塞}if(pid == 0){	signal(SIGUSR1, signal_handler);pause();}return 0;
}