Linux 消息队列的使用方法

1.概念

1. 消息队列是一种进程间通信 (IPC) 机制，允许进程之间交换数据。
2. 每个消息队列都有一个唯一的标识符 (msqid)，通过该标识符来访问消息队列。
3. 消息队列可以存储多个消息，每个消息包含消息类型和消息内容。

2. 创建消息队列

1. 使用 msgget() 函数创建消息队列。
2. msgget() 函数需要两个参数：
   int msgget(key_t key, int msgflag)
key_t key: 消息队列的键值，用于标识消息队列。
   int msgflg: 控制消息队列的创建方式，例如 IPC_CREAT 用于创建新的消息队列，IPC_EXCL 用于检查消息队列是否存在。
3. 成功创建或访问消息队列，返回正整数 或 为0消息队列标识符（ID）。
   失败返回 < 0。

---

使用 IPC_PRIVATE 创建的每个消息队列都是唯一的，即使是同一个进程多次创建，它们之间也是相互独立的，不会共享数据。

• IPC_PRIVATE 用于创建一个私有的消息队列，它只对创建它的进程可见，其他进程无法访问。并且当创建者进程终止时，该 IPC 对象也会被自动销毁。
• 当一个进程使用 IPC_PRIVATE 创建消息队列时，系统会分配一个唯一的标识符（即消息队列 ID）给它。
• 即使同一个进程多次使用 IPC_PRIVATE 创建消息队列，每次分配的标识符也会是不同的。
  因此，即使是同一个进程，每次使用 IPC_PRIVATE 创建的队列都是独立的、不同的消息队列。

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <stdio.h>
#include <stdlib.h>int main() {int msqid1 = msgget(IPC_PRIVATE, IPC_CREAT | 0666);if (msqid1 == -1) {perror("msgget 1");exit(1);}printf("Message queue ID 1: %d\n", msqid1);int msqid2 = msgget(IPC_PRIVATE, IPC_CREAT | 0666);if (msqid2 == -1) {perror("msgget 2");exit(1);}printf("Message queue ID 2: %d\n", msqid2);// msqid1 和 msqid2 不同，指向不同的消息队列if (msqid1 == msqid2) {printf("Error: msqid1 and msqid2 are the same!\n");} else {printf("msqid1 and msqid2 are different, as expected.\n");}// 删除消息队列msgctl(msqid1, IPC_RMID, NULL);msgctl(msqid2, IPC_RMID, NULL);return 0;
}

3. 发送消息

1. 使用 msgsnd() 函数发送消息到消息队列。

2. msgsnd() 函数需要四个参数：
   int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);
int msqid: 消息队列标识符。
   const void *msgp: 指向消息结构体的指针。
   size_t msgsz: 消息结构体的大小。
   int msgflg: 控制消息发送方式，例如 IPC_NOWAIT 用于非阻塞发送，填0则用于阻塞发送。

3. 返回值是 0 表示发送成功，-1 表示发送失败。

4. 接收消息

1. 使用 msgrcv() 函数从消息队列接收消息。

2. msgrcv() 函数需要五个参数：
   ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp, int msgflg);
int msqid: 消息队列标识符。
   void *msgp: 指向消息结构体的指针。
   size_t msgsz: 消息结构体的大小。
   long msgtyp: 消息类型，用于过滤接收的消息。
   int msgflg: 控制消息接收方式，例如 IPC_NOWAIT 用于非阻塞接收。填0则用于阻塞接受。

3. 返回值是接收到的消息大小，如果接收失败，则返回 -1。（若使用了 IPC_NOWAIT这个标志，在无消息时也是返回-1）

5. 消息结构体

1. 消息结构体包含消息类型 (mtype) 和消息内容 (mtext)。
2. mtype 用于区分不同类型的消息。这个类型一点是
3. mtext 可以存储任意数据。

// 定义消息结构体
struct msgbuf {long mtype;    // 消息类型，一定要放在结构体前面，而且是long类型char mtext[256]; // 消息内容
};

6. 消息队列控制（删除、获取队列状态）

1. 使用 msgctl() 函数控制消息队列。
2. msgctl() 函数需要三个参数：
   int msgctl(int msqid,int cmd,struct msqid_ds *buf)
int msqid: 消息队列标识符。
   int cmd: 控制命令，例如 IPC_STAT 用于获取消息队列状态，IPC_RMID 用于删除消息队列。
3. struct msqid_ds *buf: 指向消息队列控制结构体的指针，用于获取或设置消息队列属性。

不使用 IPC_PRIVATE 创建消息队列创建后，如果不删除，会一直存在于系统中，直到系统重启。

消息队列是否存在

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <stdio.h>
#include <errno.h>int main() {key_t key = ftok(".", 'a'); // 获取消息队列键值int msgid = msgget(key, 0); // 尝试获取现有消息队列if (msgid == -1) {if (errno == ENOENT) {printf("Message queue with key %d does not exist.\n", key);} else {perror("msgget failed");}return 1;} else {printf("Message queue with key %d exists.\n", key);return 0;}
}

7. 使用场景

进程间通信：不同进程之间传递数据。
多线程通信：不同线程之间传递数据，但需要考虑线程同步问题。

8. 注意事项

1. 消息队列的创建和使用需要权限控制。
2. 消息队列需要使用 msgctl() 函数删除，否则会导致资源泄漏。
3. 在多线程环境下使用消息队列时，需要考虑线程同步问题，避免数据竞争。
4. 注意创建消息队列的 KEY，不能用相同的KEY获取消息队列，否则将会导致消息无法传输

总结:

Linux 消息队列是一种简单易用的进程间通信机制，可以用于进程之间或线程之间交换数据。它提供了灵活的消息类型和内容存储，但需要谨慎处理线程同步问题，并注意资源释放。

使用例子

=注意：struct msgbuf 中的long mtype一定是long类型

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>#define MSG_KEY 12345// 定义消息结构体
struct msgbuf {long mtype;    // 消息类型char mtext[256]; // 消息内容
};int main() {int msqid;struct msgbuf message;// 创建消息队列msqid = msgget(MSG_KEY, IPC_CREAT | 0666);if (msqid == -1) {perror("msgget failed");exit(1);}// 进程A: 发送消息if (fork() == 0) { // 子进程printf("Process A: Sending message...\n");// 准备消息结构体message.mtype = 1;strcpy(message.mtext, "Hello from Process A!");// 发送消息if (msgsnd(msqid, &message, strlen(message.mtext) + 1, 0) == -1) {perror("msgsnd failed");exit(1);}printf("Process A: Message sent.\n");exit(0);} else { // 父进程// 等待子进程发送消息sleep(1);// 进程B: 接收消息printf("Process B: Receiving message...\n");// 接收消息if (msgrcv(msqid, &message, sizeof(message.mtext), 1, 0) == -1) {perror("msgrcv failed");exit(1);}printf("Process B: Received message: %s\n", message.mtext);}// 清理消息队列if (msgctl(msqid, IPC_RMID, NULL) == -1) {perror("msgctl failed");exit(1);}return 0;
}

#include <pthread.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>#define MSG_KEY 12345// 定义消息结构体
struct msgbuf {long mtype;    // 消息类型char mtext[256]; // 消息内容
};// 线程函数: 发送消息
void *sender(void *arg) {int msqid;struct msgbuf message;// 获取消息队列标识符msqid = *(int *)arg;while (1) {// 准备消息结构体message.mtype = 1;sprintf(message.mtext, "Message from sender thread: %ld", getpid());// 发送消息if (msgsnd(msqid, &message, strlen(message.mtext) + 1, 0) == -1) {perror("msgsnd failed");exit(1);}printf("Sender thread: Message sent: %s\n", message.mtext);sleep(1);}return NULL;
}// 线程函数: 接收消息
void *receiver(void *arg) {int msqid;struct msgbuf message;// 获取消息队列标识符msqid = *(int *)arg;while (1) {// 接收消息if (msgrcv(msqid, &message, sizeof(message.mtext), 1, 0) == -1) {perror("msgrcv failed");exit(1);}printf("Receiver thread: Received message: %s\n", message.mtext);}return NULL;
}int main() {int msqid;pthread_t sender_thread, receiver_thread;// 创建消息队列msqid = msgget(MSG_KEY, IPC_CREAT | 0666);if (msqid == -1) {perror("msgget failed");exit(1);}// 创建发送线程if (pthread_create(&sender_thread, NULL, sender, &msqid) != 0) {perror("pthread_create failed");exit(1);}// 创建接收线程if (pthread_create(&receiver_thread, NULL, receiver, &msqid) != 0) {perror("pthread_create failed");exit(1);}// 等待线程结束pthread_join(sender_thread, NULL);pthread_join(receiver_thread, NULL);// 清理消息队列if (msgctl(msqid, IPC_RMID, NULL) == -1) {perror("msgctl failed");exit(1);}return 0;
}

判断消息队列是否存在的代码

在 msgget() 函数中，将 IPC_EXCL 与 IPC_CREAT 标志位一起使用：

2int msqid = msgget(key, IPC_CREAT | IPC_EXCL | 0666);
key: 消息队列的键值。
IPC_CREAT: 如果消息队列不存在，则创建新的消息队列。
IPC_EXCL: 如果消息队列已存在，则返回错误，并设置 errno 为 EEXIST。
0666: 设置消息队列的访问权限。

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <stdio.h>
#include <stdlib.h>
#include <errno.h>int main() {int msqid;key_t key = 12345;// 第一次创建消息队列，成功msqid = msgget(key, IPC_CREAT | IPC_EXCL | 0666);if (msqid == -1) {perror("msgget failed");exit(1);}printf("Message queue created successfully.\n");// 第二次创建消息队列，失败，因为消息队列已经存在msqid = msgget(key, IPC_CREAT | IPC_EXCL | 0666);if (msqid == -1) {if (errno == EEXIST) {printf("Message queue already exists.\n");} else {perror("msgget failed");exit(1);}}return 0;
}

解释:

• 第一次调用 msgget() 时，使用 IPC_CREAT | IPC_EXCL 标志位，成功创建消息队列。
• 第二次调用 msgget() 时，使用相同的键值，由于消息队列已存在，IPC_EXCL 标志位会使 msgget() 返回错误，errno 为 EEXIST。

---

使用场景:

• 当需要确保创建的消息队列是新的，并且不允许重复创建时，使用 IPC_EXCL 标志位。
• 在多进程或多线程环境中，使用 IPC_EXCL 可以防止多个进程或线程同时创建相同的消息队列。

获取队列空间大小

步骤：

1. 获取消息队列控制结构体：
   使用 msgctl() 函数，并设置 cmd 为 IPC_STAT，将消息队列控制结构体指针作为第三个参数传入。
   

2. 获取 msg_qbytes 字段：
   msqid_ds 结构体中包含消息队列的各种属性，包括 msg_qbytes 字段，它表示消息队列的最大容量（字节）。

3. 计算队列大小：
   msg_qbytes 代表的是消息队列的总容量，如果需要计算可以容纳的消息数量，需要考虑每个消息的大小。
   int max_messages = msqid_ds.msg_qbytes / sizeof(struct msgbuf); // 假设 struct msgbuf是你的消息结构体

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <stdio.h>
#include <stdlib.h>int main() {int msqid;key_t key = 12345;// 创建消息队列msqid = msgget(key, IPC_CREAT | 0666);if (msqid == -1) {perror("msgget failed");exit(1);}// 获取消息队列控制结构体struct msqid_ds msqid_ds;if (msgctl(msqid, IPC_STAT, &msqid_ds) == -1) {perror("msgctl failed");exit(1);}// 打印消息队列最大容量printf("Message queue size: %ld bytes\n", msqid_ds.msg_qbytes);// 计算可以容纳的消息数量（假设消息结构体大小为 100 字节）int max_messages = msqid_ds.msg_qbytes / 100;printf("Maximum number of messages: %d\n", max_messages);// 删除消息队列if (msgctl(msqid, IPC_RMID, NULL) == -1) {perror("msgctl failed");exit(1);}return 0;
}