IPC通信之消息队列 Message Queue

今天将使用 POSIX Message Queue (POSIX 消息队列)。

相比于管道（字节流）和共享内存（原始内存块），消息队列最大的优点是“有边界”和“有优先级”。

实验设计

Queue: 创建一个名为 /my_mq 的队列。
two.c (发送者): 发送带有优先级的消息。比如发送一条紧急消息（优先级 10）和一条普通消息（优先级 1）。
one.c (接收者): 从队列里取消息。你会发现，即使普通消息先发，紧急消息也会被先取出来！

代码

one.c

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <mqueue.h>
#include <errno.h>

#define MQ_NAME "/my_mq"
#define MAX_SIZE 1024

int main() {
    mqd_t mq;
    struct mq_attr attr;
    char buffer[MAX_SIZE + 1];
    unsigned int prio;

    // 1. 设置队列属性
    attr.mq_flags = 0;
    attr.mq_maxmsg = 10;      // 队列最多容纳 10 条消息
    attr.mq_msgsize = MAX_SIZE; // 每条消息最大 1024 字节
    attr.mq_curmsgs = 0;

    // 2. 创建消息队列
    // O_CREAT: 不存在则创建
    // O_RDONLY: 只读
    mq = mq_open(MQ_NAME, O_CREAT | O_RDONLY, 0644, &attr);
    if (mq == (mqd_t)-1) {
        perror("mq_open");
        return 1;
    }

    printf("=== 接收者 ONE (PID: %d) ===\n", getpid());
    printf("消息队列已打开: %s\n", MQ_NAME);
    printf("等待消息中...\n");

    while(1) {
        // 3. 接收消息
        // mq_receive 会阻塞，直到有消息到来
        // 注意：它会自动把优先级最高的先拿出来！
        ssize_t bytes_read = mq_receive(mq, buffer, MAX_SIZE, &prio);
        
        if (bytes_read >= 0) {
            buffer[bytes_read] = '\0'; // 加上字符串结束符
            printf("[ONE] 收到消息 (优先级 %u): %s\n", prio, buffer);
            
            if (prio >= 10) {
                printf("      !!! 这是一个紧急消息 !!!\n");
            }
        } else {
            perror("mq_receive");
            break;
        }
    }

    // 清理
    mq_close(mq);
    mq_unlink(MQ_NAME);
    return 0;
}

two.c

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <mqueue.h>
#include <string.h>

#define MQ_NAME "/my_mq"

int main(int argc, char *argv[]) {
    mqd_t mq;
    char buffer[1024];
    unsigned int prio = 1; // 默认优先级

    if (argc < 2) {
        printf("使用方法: %s <消息内容> [优先级(0-31)]\n", argv[0]);
        return 1;
    }

    // 获取消息内容
    strcpy(buffer, argv[1]);
    
    // 获取优先级 (如果有的话)
    if (argc >= 3) {
        prio = atoi(argv[2]);
    }

    // 1. 打开消息队列
    // O_WRONLY: 只写
    mq = mq_open(MQ_NAME, O_WRONLY);
    if (mq == (mqd_t)-1) {
        perror("mq_open (可能 one.c 还没运行?)");
        return 1;
    }

    printf("=== 发送者 TWO ===\n");
    printf("正在发送: \"%s\" (优先级: %u)...\n", buffer, prio);

    // 2. 发送消息
    if (mq_send(mq, buffer, strlen(buffer), prio) == -1) {
        perror("mq_send");
        return 1;
    }

    printf("发送成功！\n");

    mq_close(mq);
    return 0;
}

实验步骤

接受修改。
编译（可能需要链接 -lrt，视系统而定）：
```
gcc one.c -o process_one
gcc two.c -o process_two
```
运行接收者 (one)：
```
./process_one
```
它会阻塞等待。
运行发送者 (two)（在另一个终端）：
- 发送普通消息：
```
./process_two "你好，这是一条普通消息" 1
```
- 发送紧急消息：
```
./process_two "着火啦！快跑！" 20
```
- 进阶玩法：先杀掉 process_one，然后疯狂运行 process_two 发送一堆不同优先级的消息（有些是 1，有些是 30）。然后再启动 process_one。你会发现，它不是按时间顺序收的，而是先把所有高优先级的收完了，才去收低优先级的！这就是消息队列的强大之处。

POSIX 消息队列的位置取决于操作系统。

Linux 系统

在 Linux 上，消息队列通常被挂载在一个特殊的虚拟文件系统中：
/dev/mqueue/

你可以通过以下命令查看：

ls -l /dev/mqueue/

你应该能看到一个名为 my_mq 的文件。你甚至可以用 cat /dev/mqueue/my_mq 查看它的一些元数据（如队列当前有多少条消息），但不能直接看内容。

注意：如果 /dev/mqueue 不存在，你需要手动挂载它（通常系统会自动挂载）：
mount -t mqueue none /dev/mqueue

macOS 系统

macOS 对 POSIX IPC 的实现比较封闭。它不把消息队列暴露在文件系统中。
你无法在 Finder 或终端的 /dev 目录下找到它。它完全存在于内核内存中，只有通过 mq_open 等 API 才能访问。

总结

Linux: 去 /dev/mqueue/ 找。
macOS: 找不到，它在内核的隐秘角落里。

解读 cat /dev/mqueue/my_mq

发了三次 "hi 3"、"hi 2"、"hi 1"。

$ cat my_mq 
QSIZE:12         NOTIFY:0     SIGNO:0     NOTIFY_PID:0

这是 Linux 下查看 /dev/mqueue/my_mq 内容时的典型输出。它展示了该消息队列当前的状态元数据。

我们逐项解读：

1. `QSIZE:12` (Queue Size)

含义: 当前队列中堆积的数据总字节数是 12 字节。
推测: 这意味着你往里面发了一些消息，但还没有被 process_one 取走（或者取走了一部分）。
注意: 这不是消息的条数，而是总大小。如果你发了 3 条消息，每条内容是 "Hi"（2字节+结束符），那么可能就是这个大小。

2. `NOTIFY:0` (Notification Mode)

含义: 0 表示当前没有进程注册了“异步通知”。
背景: 消息队列支持一种高级玩法——“当有消息来时，给我发个信号或者创建一个线程”。如果这个值是 1 或 2，说明有人订阅了这个事件。

3. `SIGNO:0` (Signal Number)

含义: 如果开启了通知模式，内核该发送哪个信号（比如 SIGUSR1）给订阅者。这里是 0，因为没开启通知。

4. `NOTIFY_PID:0` (Notification PID)

含义: 那个订阅了通知的进程 PID 是多少。这里是 0，说明没人订阅。

隐藏的信息（不在这个 cat 输出里，但在 `mq_attr` 结构体里）

虽然 cat 只显示了这些，但其实还有两个关键指标：

mq_curmsgs: 当前有多少条消息？（这个最重要，可惜 cat 没直接显示，但可以通过 QSIZE 估算）
mq_maxmsg: 队列最多能存多少条消息？（如果满了，two.c 再发就会阻塞或报错）

结论

你的队列里目前有货（12字节）。这意味着如果你现在启动 process_one，它会立即读到这 12 字节的数据。

电脑关机重启数据还在吗？

不在了。

无论是 Signal、Pipe、Shared Memory 还是 Message Queue，它们的生命周期都不超过内核的生命周期。

为什么？

因为它们的数据本质上都是存在 RAM (内存) 里的。

/dev/mqueue 是虚拟文件系统。
/dev/shm 是 tmpfs (内存文件系统)。
管道缓冲区更是纯粹的内核内存对象。

一旦断电（关机），内存数据瞬间清空，这些 IPC 对象也就随之灰飞烟灭。

如果我需要重启后数据还在？

那你不能用标准的 OS IPC，你需要持久化 (Persistence)：

普通文件: 写到硬盘上的 /var/log/myapp.txt 或 config.json。
数据库: 用 SQLite, MySQL, Redis (开启 AOF/RDB)。
中间件: 使用 RabbitMQ, Kafka 等消息队列中间件，它们支持把消息存盘。

唯一的例外（命名管道 FIFO）

如果你创建了一个 命名管道 (FIFO) 文件（比如 /tmp/my_pipe）：

重启后，文件本身（那个门牌号）可能还在（取决于它所在的目录是不是 /tmp 这种临时目录，如果是 /home/user/my_pipe 就肯定在）。
但是！管道里的数据（刚才发的那句 "Hello"）绝对没了。那个管道变成了空的。

IPC通信之消息队列 Message Queue