【并发程序设计】14.消息队列

14.消息队列

消息队列（Message Queue）是一种通信机制，用于在分布式系统中传递和管理消息的队列型数据结构。

消息队列通常是一个先进先出（FIFO）的数据结构，它允许多个进程或线程之间以异步方式进行通信。
它可以被看作是一个系统内核中的内部链表，其中发送进程将消息添加到队列中，接收进程从队列中读取消息进行处理。
这种通信机制传递的数据通常是结构化的，而不是简单的字节流。

在这里插入图片描述

消息队列使用步骤

发送端：
1. 申请Key ftok
2. 打开/创建消息队列 msgget
3. 向消息队列发送消息 msgsnd
接收端：
1. 打开/创建消息队列 msgget
2. 从消息队列接收消息 msgrcv
3. 控制（删除）消息队列 msgctl

发送端

1.申请Key

ftok函数

原型：key_t ftok(const char *pathname, int proj_id);
功能：生成一个唯一的键值，通常用于创建共享内存或消息队列等
参数：
pathname：一个已经存在的文件路径，通常是程序中已经打开的文件。
proj_id：一个整数，用于与pathname组合生成唯一的键值。

返回值：
成功，返回一个key_t类型的键值，用于后续的系统调用（如shmget、msgget等）
失败，返回-1

2.打开/创建消息队列

msgget函数

原型：int msgget(key_t key, int msgflg);
功能：创建或打开一个消息队列
参数：
key：一个唯一的键值，通常由ftok函数生成。
msgflg：消息队列的访问权限和创建标志,IPC_CREAT、IPC_EXCL、IPC_NOWAIT等
IPC_CREAT
若消息队列不存在，则创建一个新的消息队列。

IPC_EXCL
当与 IPC_CREAT 同时使用时，如果已经存在与 key 相关联的消息队列，msgget 会失败并返回 -1。
如果没有同时指定 IPC_CREAT，IPC_EXCL 将被忽略。

IPC_NOWAIT
如果消息队列不能立即创建（例如，由于资源限制），msgget 会立即返回 -1，而不是等待直到可以创建为止。
如果没有指定 IPC_NOWAIT，系统会等待直到可以创建消息队列为止。

IPC_PRIVATE
在 Linux 中，IPC_PRIVATE 是一种特殊的键值，它允许不使用 ftok 来生成唯一的 key。
使用 IPC_PRIVATE 作为 key 时，系统会为调用者分配一个私有的、唯一的消息队列标识符。

IPC_RMID
这是一个较新的选项，用于删除所有关联的消息队列，而不仅仅是与给定 key 关联的那个。

0666 或 0777 等权限位
这些数字代表消息队列的访问权限。通常，它们被设置为八进制数，其中每个数字对应于用户、组和其他人的读/写权限。

返回值：
成功，返回消息队列的标识符（非负整数）
失败，返回-1

向消息队列发送消息 msgsnd函数
原型：int msgsnd(int msqid, const struct msgbuf *msgp, int msgsz, int msgflg);
功能：向消息队列发送一个消息
参数：
msqid：消息队列的标识符，通常由msgget函数返回。
msgp：指向要发送的消息结构的指针，该结构包含消息类型和消息正文。
msgsz：消息的大小（以字节为单位）。
msgflg：指定消息发送的标志，可以是IPC_NOWAIT等,同msgget函数中的msgflg参数。

返回值：
成功，返回0
失败，返回-1

示例

发送一条消息到队列

#include <stdio.h> 
#include <sys/ipc.h> 
#include <sys/msg.h> 
#include <string.h> typedef struct // 定义一个结构体类型
{long msg_type; // 消息类型，必须是long型char buf[128]; // 消息内容缓冲区，用于存储消息内容
}msgT;    #define MSGLEN  (sizeof(msgT)-sizeof(long)) // 定义消息长度宏，减去长整型变量的大小int main() // 主函数
{key_t key; // 定义一个key_t类型的变量key，用于存储生成的键值int msgid; // 定义一个整型变量msgid，用于存储消息队列的标识符int ret;   // 定义一个整型变量ret，用于存储函数调用的返回值msgT msg;  // 定义一个msgT类型的结构体变量msg，用于存储要发送的消息//1.生成一个键值key = ftok(".",100); // "."表示当前目录，100是一个任意的整数if(key<0) // 如果生成键值失败{perror("ftok"); // 打印错误信息并返回0return 0;}//2.函数创建msgid = msgget(key,IPC_CREAT|0666); // key为键值，IPC_CREAT表示如果不存在则创建，0666表示权限设置为可读写if(msgid<0) // 如果获取消息队列失败{perror("msgget"); // 打印错误信息并返回0return 0;}//3.发送消息msg.msg_type = 1; // 设置消息类型为1strcpy(msg.buf,"this msg type 1"); // 将消息内容复制到buf中ret = msgsnd(msgid,&msg,MSGLEN,0);//0表示不等待if(ret<0) // 如果发送消息失败{perror("msgsnd"); // 打印错误信息并返回0return 0;}
}

运行程序后，在终端输入ipcs -qc查看消息队列

在这里插入图片描述

这就是刚刚发送到消息队列的信息

接收端

打开/创建消息队列 msgget函数（发送端用过）

从消息队列接收消息

msgrcv函数

原型：ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp, int msgflg);
功能：从消息队列中接收一个消息
参数：
msqid：消息队列的标识符，通常由msgget函数返回。
msgp：指向要接收的消息结构的指针，该结构包含消息类型和消息正文。
msgsz：消息的大小（以字节为单位）。
msgtyp：指定要接收的消息类型。
msgtyp=0：收到的第一条消息，任意类型。
msgtyp>0：收到的第一条 msg_typ类型的消息。
msgtyp<0：接收类型等于或者小于msgtyp绝对值的第一个消息。

msgflg：
0：阻塞式接收消息
IPC_NOWAIT：如果没有返回条件的消息调用立即返回，此时错误码为ENOMSG
MSG_EXCEPT：与msgtyp配合使用返回队列中第一个类型不为msgtyp的消息

返回值：
成功时返回实际接收到的消息大小（以字节为单位）
失败时返回-1

控制（删除）消息队列 msgctl函数
原型：int msgctl(int msqid, int cmd, struct msqid_ds *buf);
功能：控制消息队列，包括获取和设置消息队列的属性
参数：
msqid：消息队列的标识符，通常由msgget函数返回。
cmd：指定要执行的命令，可以是IPC_STAT、IPC_SET等。
buf：指向一个msqid_ds结构体的指针，用于存储或设置消息队列的属性。

返回值：
成功时返回0
失败时返回-1

示例

接收消息队列的一条消息

#include <stdio.h> 
#include <sys/ipc.h> 
#include <sys/msg.h> 
#include <string.h> // 定义消息结构体
typedef struct
{long msg_type; // 消息类型char buf[128]; // 消息内容缓冲区
}msgT;    #define MSGLEN  (sizeof(msgT)-sizeof(long)) // 计算消息长度，减去长整型成员的长度
int main()
{int msgid; // 消息队列IDkey_t key; // 消息队列关键字msgT msg; // 消息结构体实例int ret; // 返回值key = ftok(".",100); // 生成消息队列关键字if(key<0){perror("ftok"); // 如果生成失败，打印错误信息return 0;}// 1.创建或获取消息队列IDmsgid = msgget(key,IPC_CREAT|0666); if(msgid<0){perror("msgget"); // 如果获取失败，打印错误信息return 0;}// 2.从消息队列中接收消息ret = msgrcv(msgid,&msg,MSGLEN,0,0); if(ret<0){perror("msgrcv"); // 如果接收失败，打印错误信息return 0;}printf("receiv msg type=%d,buf=%s\n",(int)msg.msg_type,msg.buf); // 打印接收到的消息类型和内容// 3.删除消息队列ret = msgctl(msgid,IPC_RMID,NULL);if(ret<0){perror("msgctl"); // 如果删除失败，打印错误信息return 0;}
}