Obsidian-Knowledge-Base/X-Knowledge-Base/raw/Joplin/Linux+C+C++技术体系梳理/2. Linux系统编程/21. 消息队列.md

消息队列 (Message Queue)

概念

消息队列是由内核维护的消息链表。消息队列允许进程以消息的形式进行异步通信，消息带有类型标识，接收方可以根据类型选择性地接收消息。

核心特点：

• 消息存储在内核中，进程退出后消息仍存在（生命周期由内核管理）
• 支持消息类型过滤，接收方可按类型选择性读取
• 异步通信：发送方无需等待接收方读取
• 适合嵌入式场景中的事件通知和进程间异步消息传递

System V 消息队列

关键API

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

// 创建/获取消息队列
int msgget(key_t key, int msgflg);

// 发送消息
int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);

// 接收消息
ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp, int msgflg);

// 控制消息队列
int msgctl(int msqid, int cmd, struct msqid_ds *buf);

消息结构体

struct msgbuf {
    long mtype;       // 消息类型，必须 > 0
    char mtext[1];    // 消息数据（柔性数组，实际大小由用户定义）
};

msgflg 标志

标志	说明
IPC_CREAT	不存在则创建
IPC_EXCL	与 IPC_CREAT 配合，已存在则报错
0666	权限位
IPC_NOWAIT	非阻塞模式，队列满/空时立即返回错误

msgrcv 的 msgtyp 参数

msgtyp 值	行为
> 0	接收指定类型的消息
0	接收队列中第一条消息（任意类型）
< 0	接收类型 ≤

POSIX 消息队列

关键API

#include <mqueue.h>

// 打开/创建消息队列
mqd_t mq_open(const char *name, int oflag, ...);

// 发送消息
int mq_send(mqd_t mqdes, const char *msg_ptr, size_t msg_len, unsigned int msg_prio);

// 接收消息
ssize_t mq_receive(mqd_t mqdes, char *msg_ptr, size_t msg_len, unsigned int *msg_prio);

// 关闭消息队列
int mq_close(mqd_t mqdes);

// 删除消息队列
int mq_unlink(const char *name);

POSIX vs System V：

• POSIX 使用命名队列（/mq_name），更直观
• POSIX 支持优先级
• POSIX 有 mq_notify 异步通知机制
• POSIX 是可移植的标准接口

完整代码示例：System V 消息队列

// mq_example.c
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <unistd.h>
#include <sys/wait.h>

#define MSG_KEY 0x1234
#define MSG_TYPE_DATA 1
#define MSG_TYPE_CMD  2

struct msgbuf {
    long mtype;
    char mtext[256];
};

// 发送进程
void sender(int msqid) {
    struct msgbuf msg;

    // 发送数据消息
    msg.mtype = MSG_TYPE_DATA;
    snprintf(msg.mtext, sizeof(msg.mtext), "Hello from sender, pid=%d", getpid());
    if (msgsnd(msqid, &msg, strlen(msg.mtext) + 1, 0) == -1) {
        perror("msgsnd");
        exit(EXIT_FAILURE);
    }
    printf("[Sender] Sent DATA: %s\n", msg.mtext);

    // 发送命令消息
    msg.mtype = MSG_TYPE_CMD;
    snprintf(msg.mtext, sizeof(msg.mtext), "CMD:SHUTDOWN");
    if (msgsnd(msqid, &msg, strlen(msg.mtext) + 1, 0) == -1) {
        perror("msgsnd");
        exit(EXIT_FAILURE);
    }
    printf("[Sender] Sent CMD: %s\n", msg.mtext);
}

// 接收进程
void receiver(int msqid) {
    struct msgbuf msg;

    // 只接收数据消息
    if (msgrcv(msqid, &msg, sizeof(msg.mtext), MSG_TYPE_DATA, 0) == -1) {
        perror("msgrcv");
        exit(EXIT_FAILURE);
    }
    printf("[Receiver] Got DATA: %s\n", msg.mtext);

    // 接收命令消息
    if (msgrcv(msqid, &msg, sizeof(msg.mtext), MSG_TYPE_CMD, 0) == -1) {
        perror("msgrcv");
        exit(EXIT_FAILURE);
    }
    printf("[Receiver] Got CMD: %s\n", msg.mtext);
}

int main(void) {
    int msqid;

    // 创建消息队列
    msqid = msgget(MSG_KEY, IPC_CREAT | 0666);
    if (msqid == -1) {
        perror("msgget");
        exit(EXIT_FAILURE);
    }
    printf("Message queue created, id=%d\n", msqid);

    pid_t pid = fork();
    if (pid == -1) {
        perror("fork");
        exit(EXIT_FAILURE);
    }

    if (pid == 0) {
        // 子进程：接收
        receiver(msqid);
        exit(EXIT_SUCCESS);
    } else {
        // 父进程：发送
        sender(msqid);
        wait(NULL);
    }

    // 删除消息队列
    msgctl(msqid, IPC_RMID, NULL);
    printf("Message queue removed\n");

    return 0;
}

编译与运行：

gcc -o mq_example mq_example.c
./mq_example

注意事项

1. 消息大小限制：单条消息最大通常为 8192 字节（可通过 /proc/sys/kernel/msgmnb 调整）
2. 消息队列总容量：系统级限制，可通过 ipcs -q 查看
3. 内存占用：消息存储在内核内存中，大量消息会消耗内核内存
4. 权限控制：消息队列有独立的权限位，注意安全配置
5. 清理：进程退出后消息队列不会自动删除，需主动调用 msgctl(IPC_RMID) 或 ipcrm
6. 字节序：嵌入式场景注意大小端一致性

面试要点

1. 消息队列 vs 管道：消息队列支持消息类型过滤、异步通信、不依赖父子进程关系
2. 消息队列 vs 共享内存：消息队列有内核同步开销，但使用简单；共享内存最快但需自行同步
3. 生命周期：System V 消息队列是内核持久的，进程退出后仍存在
4. POSIX vs System V：POSIX 更现代、可移植，支持异步通知（mq_notify）
5. 典型应用场景：嵌入式系统中的事件驱动架构、日志系统、命令分发