第3章：消息队列（Message Queue）：System V消息队列与POSIX消息队列对比
消息队列这东西，说白了就是进程间传递数据的一块「公共黑板」。一个进程往上面贴消息，另一个进程来取。我在早年做嵌入式网关项目时，就靠它来协调多个传感器采集进程和数据处理进程之间的通信。今天咱们就聊聊两种主流的消息队列——System V和POSIX，看看它们到底有啥区别，怎么用，以及实战中怎么搭一个多生产者-单消费者的模型。
3.1 System V消息队列 vs POSIX消息队列
这两种队列，名字听着像两兄弟，但脾气秉性差不少。我刚开始接触时也犯过混，后来踩了几个坑才摸清楚。
| 对比维度 | System V消息队列 | POSIX消息队列 |
|---|---|---|
| 标识方式 | 使用 key（键值）和 msgid（队列ID） | 使用名字（以 / 开头），类似文件路径 |
| 创建/打开 | msgget()，需要指定 key 和标志 | mq_open()，直接传名字和标志 |
| 消息类型 | 支持按消息类型（长整型）接收 | 不支持类型，只能按优先级接收 |
| 最大消息数 | 系统级限制，通常较小 | 可配置，通常更灵活 |
| 异步通知 | 不支持，只能阻塞或轮询 | 支持 mq_notify() 注册信号或线程通知 |
| 清理方式 | 需要手动 ipcrm 或程序删除 | 进程退出时自动清理（引用计数） |
我个人习惯是：如果项目需要按消息类型精确分发，我会选System V。如果追求易用性和异步通知，POSIX更顺手。但要注意，POSIX队列在老旧内核上可能不支持，我在一个CentOS 6的项目里就吃过这个亏。
3.2 消息队列的创建与销毁
创建队列其实不复杂，但细节决定成败。咱们先看System V的创建方式：
#include <sys/msg.h>
// 创建或获取一个消息队列
int msgid = msgget(IPC_PRIVATE, IPC_CREAT | 0666);
// 或者用固定 key
key_t key = ftok("/tmp/msgqueue", 'A');
int msgid = msgget(key, IPC_CREAT | 0666);
// 销毁队列
msgctl(msgid, IPC_RMID, NULL);
这里有个坑：ftok() 生成的 key 可能冲突。我曾经在一个集群环境中，两个服务用了同一个路径和项目ID，结果消息串了，排查了一整天。所以建议用 IPC_PRIVATE 配合进程间传递 msgid，或者用 ftok() 时选一个独特的路径。
再看POSIX的创建：
#include <mqueue.h>
// 创建或打开
struct mq_attr attr;
attr.mq_flags = 0;
attr.mq_maxmsg = 10; // 最大消息数
attr.mq_msgsize = 1024; // 每条消息最大字节
attr.mq_curmsgs = 0;
mqd_t mqd = mq_open("/myqueue", O_CREAT | O_RDWR, 0666, &attr);
// 关闭（不是销毁）
mq_close(mqd);
// 销毁队列
mq_unlink("/myqueue");
注意：POSIX队列的名字必须以
/ 开头，且长度有限制（通常不超过255字符）。我见过有人忘了加斜杠，结果 mq_open 一直返回失败，还以为是权限问题。
3.3 实战：多生产者-单消费者模型
这个模型在实际中太常见了。比如日志收集系统，多个工作线程产生日志，一个写入线程负责落盘。咱们用POSIX消息队列来实现，因为它支持异步通知，写起来更优雅。
生产者代码（多个进程/线程可同时运行）
#include <mqueue.h>
#include <string.h>
#include <unistd.h>
void producer(const char *data) {
mqd_t mqd = mq_open("/logqueue", O_WRONLY);
if (mqd == (mqd_t)-1) {
perror("mq_open");
return;
}
// 发送消息
if (mq_send(mqd, data, strlen(data) + 1, 0) == -1) {
perror("mq_send");
}
mq_close(mqd);
}
消费者代码（单线程循环接收）
#include <mqueue.h>
#include <stdio.h>
#include <signal.h>
mqd_t mqd;
void handle_msg(union sigval sv) {
char buf[1024];
unsigned int prio;
// 非阻塞接收
ssize_t n = mq_receive(mqd, buf, sizeof(buf), &prio);
if (n > 0) {
printf("收到消息: %s (优先级: %u)\n", buf, prio);
}
}
int main() {
struct mq_attr attr;
attr.mq_maxmsg = 20;
attr.mq_msgsize = 1024;
mqd = mq_open("/logqueue", O_CREAT | O_RDONLY, 0666, &attr);
// 注册异步通知
struct sigevent sev;
sev.sigev_notify = SIGEV_THREAD;
sev.sigev_notify_function = handle_msg;
sev.sigev_notify_attributes = NULL;
mq_notify(mqd, &sev);
// 主线程做其他事，或者 sleep
pause();
mq_close(mqd);
mq_unlink("/logqueue");
return 0;
}
小技巧：如果消费者处理不过来，可以设置队列的最大消息数来反压生产者。我在一个实时数据采集项目中，把 mq_maxmsg 设为100，当队列满了，生产者 mq_send 会阻塞，自然就限流了。
3.4 避坑指南
做消息队列开发，有几个坑我不得不提：
- 消息大小限制：System V 的 msgmax 通常只有 8KB，POSIX 可以设大一些，但也不能超过内核限制。我曾经传一个 10KB 的结构体，结果 msgrcv 一直返回 E2BIG，查了半天才发现是大小超了。
- 队列残留：System V 队列不会随进程退出而删除。我有个同事，开发时频繁创建队列但不清理，最后系统报 "No space left on device"，其实是消息队列占满了内核内存。
- 优先级反转：POSIX 队列按优先级接收，如果高优先级消息不断涌入，低优先级消息可能永远得不到处理。我在一个监控系统里就遇到过，告警消息优先级高，把普通日志消息饿死了。
3.5 选型建议
说了这么多，到底选哪个？我个人建议：
- 如果你需要按消息类型精确路由（比如类型1给A进程，类型2给B进程），选 System V。
- 如果你需要异步通知、自动清理、或者跨平台（POSIX 标准更统一），选 POSIX。
- 如果性能要求极高，两个都差不多，但 System V 在极端情况下略快一点点（少一层抽象）。
嗯，消息队列就聊到这儿。下一章咱们聊聊共享内存，那又是另一番天地了。