如果你已经读过我们讲过的那三种经典 IPC——匿名管道、命名管道、还有共享内存——你大概已经习惯了这样一个画面：进程之间要传数据，就得在某个"中介"上做文章。匿名管道是内核里的一条字节流，命名管道把它挂到了文件系统上，共享内存干脆把一块内存直接亮出来让大家一起用。今天我们要认识的 System V 消息队列（System V Message Queue），是另一类格外有趣的"中介"。
先说一个最戳人的点：你在别的教程里可能见过"消息队列"这个词，但那些大多是"消息中间件""RabbitMQ""Kafka"这类应用层的分布式消息系统，跟内核里这个以毫秒计、进程间直接秃噜数据的消息队列完全是两回事。我们今天说的，是操作系统里一个真正的、常驻内核的通信设施。它跟管道最不一样的地方在于：管道里流动的是没有边界的字节流，进程没法区分"这坨字节是哪条消息"；而消息队列流动的，是一个一个有"类型"、有"长度边界"的独立消息块。这个"类型"就是后面要重点讲的 mtype，它让接收方可以"点名"要哪种消息，这是管道给不了的。
我们把这一课拆开来看。你要掌握的东西其实就三板斧：
- 四个接口函数：
msgget（创建/获取队列）、msgsnd（发消息）、msgrcv（收消息）、msgctl（控制/删除队列）。把这四个用得滚瓜烂熟，消息队列的"能用"你就到手了。 - 一套阻塞与类型规则：什么时候
msgsnd会阻塞、msgrcv又是"按类型"怎么个取法。这是消息队列跟管道本质不同的地方，也是最容易出 bug 的地方。 - 一种设计模式：消息进了队列、被进程读出来之后，往往要经过"格式加工 → 落盘 → 超限备份"这样一串前后相继的处理。把这一串处理用"责任链模式"（Chain of Responsibility Pattern）组织起来，代码会干净、可扩展得多。这也是本次加餐真正想给你的"软技能"。
老规矩，所有代码我都尽量用完整、能直接 gcc 编译的小程序，并且逐行注释。你别光看，敲一遍再运行一遍，很多坑自己就浮出来了。
消息队列到底是什么
一句话定义：消息队列是内核维护的一块"带类型、按先后排队的消息集合"，它允许一个进程向它写入一条消息，允许另一个进程按类型把它读出来。 你可以把它想象成一个"邮箱"：往里面投信的人（发送进程）不用等收信人（接收进程）在不在场，信先搁在邮箱里；取信的人按"信的编号"（类型）来挑自己想要的。中间这封信就一直躺在内核里，谁也碰不到，直到有人把它取走。
为此，内核必须同时为我们维护下面这几样东西，缺一不可：
- 一个有名字的"邮箱"本身——在 System V 里用
key（IPC key）来命名，一个key全局唯一地对应一个队列； - 一堆排好队的消息——每条消息含一个
long型的类型mtype，和一段正文mtext，正文最长不能超过MSGMAX； - 一套收纳上限——
MSGMNB限制单个队列最多能装下的总字节数，MSGMNI限制整个系统最多允许多少个队列。
代码画一下内核认识中的"一条消息"长什么样（这也是我们写程序时用的 msgbuf 结构的模板）：
struct msgbuf {
long mtype; /* 消息类型，必须是 > 0 的正整数 */
char mtext[1]; /* 消息正文，长度由调用时传入的 msgsz 决定 */
};你看，这个结构其实"不完整"——mtext[1] 只是占了个位置，真正正文有多少字节，是在调用 msgsnd/msgrcv 时由 msgsz 这个参数告诉内核的。内核拿到的是一条" long 类型 + 变长正文"的二进制片断。所以消息队列传的是"内核眼里的一段有长度的数据"，不是字符串，更不是 C 字符串——它不在乎你正文里有没有 '\0'，它只认真数。这一点在后面"长度的单位"那一节会坑死不少人，我们到时候细说。
值得先记住的还有一个反直觉的结论：消息队列里消息是"先进先出"的吗？不全对。 真实规则是：msgrcv 取消息时，既可以顺着队头按顺序取（msgtyp = 0 就是"取队头第一条"），也可以跳过前面的、专门取某一种类型（msgtyp > 0）。所以它是一个"排队 + 挑选"并存的容器。我们专门开一节讲这四种取法。
你可能会问：这么个东西，跟管道、共享内存比，凭什么存在？把三者放一桌对比，各自的定位就清楚了：
| 维度 | 匿名管道 | 命名管道 | 消息队列 | 共享内存 |
|---|---|---|---|---|
| 是否有血缘关系要求 | 必须有父子关系 | 无，任意进程 | 无，任意进程 | 无，任意进程 |
| 传输单位 | 无边界字节流 | 无边界字节流 | 有界的消息块 | 无界内存区 |
| 是否带类型 | 不带 | 不带 | 带 mtype | 不带 |
| 是否支持按类型取 | 不支持 | 不支持 | 支持 | 不支持 |
| 生命周期 | 随进程 | 随文件（需删除） | 随内核（需手动清理） | 随内核（需手动清理） |
看到最后一行没有——消息队列和共享内存一样，生命周期跟着内核走。意思很残酷：哪怕所有进程都退出了，消息队列还赖在内核里占着资源，只有你用 ipcrm 或 msgctl(..., IPC_RMID, ...) 显式删除，它才消失。这条 "你不删它不散" 的铁律，是无数初学消息队列的人内存泄漏噩梦的来源，我们专门有一节讲。
消息队列的通信形式
消息队列天然支持全双工通信。为什么？因为它不是像匿名管道那种"一头只读、一头只写"的定向通道，而是一个共享的、两边都能读能写的公共邮箱：进程 A 既能往里写、也能往回收，进程 B 同样。你要是让 A 发给 B 也成立、让 B 发给 A 也成立——双方各自往队列里 msgsnd、各自 msgrcv 取走自己想要类型的消息，没有任何方向限制。
/* 全双工示意：
* 取不同类型就能天然做到"我不抢你的消息" */
#define TYPE_TO_SERVER 1 /* A 发的，B 收 */
#define TYPE_TO_CLIENT 2 /* B 发的，A 收 */一般一个客户端-服务端的小项目就这么玩：client 发的消息用类型 1，server 只收类型 1 的；反过来 server 要回消息给 client，就用类型 2，client 只收类型 2。因为类型"点名"了接收方，两边共用同一个队列也不会互相搅浑。这正是消息类型存在的最大价值之一。
内核给每个 IPC 对象上的"户口本"：ipc_perm
凡是 System V 的 IPC 对象（消息队列、共享内存、信号量），内核都会给它们登记一份"户口本"，这个结构就叫 struct ipc_perm，定义在 sys/ipc.h 里。你可以把它理解成"这个内核对象归谁管、别的进程能不能碰"的权限信息。先看它长什么样：
struct ipc_perm {
key_t key; /* 创建者传给 xxxget(2) 的 key，即"对象的名字" */
uid_t uid; /* 当前属主的有效用户 ID（Effective UID） */
gid_t gid; /* 当前属主的有效组 ID */
uid_t cuid; /* 创建者的有效用户 ID，创建后一般不改 */
gid_t cgid; /* 创建者的有效组 ID */
unsigned short mode; /* 权限位，类似文件权限 */
unsigned short seq; /* 对象序号，用于预防"对象号被重用"的问题 */
};这里面大部分字段看着眼熟，uid/gid/mode 简直就像 struct stat 里的那一套。mode 用的就是九个权限位，跟创建普通文件的 0666、0600 这些写法一脉相承，msgget(..., IPC_CREAT | IPC_EXCL | 0666) 里最后的 0666 就是填到这里来的。它决定"别的用户能不能往这个队列里发消息/收消息"。
有一个细节值得一提：cuid/cgid（创建者的 UID/GID）是"出生证明"，创建后很少变；而 uid/gid（当前属主）可以被修改——这跟文件里的 chown 语义类似。我们平时写例程时，绝大多数情况是同一个用户自己跟自己通信，不太会去折腾这几个权限字段，但你要知道 msgctl 的 IPC_SET 可以改它。
消息队列在内核里的"房间布局"：msqid_ds
光有户口本还不够，一个消息队列具体"装了多少消息、还能装多少字节、谁最后动过它"，内核需要一份更详细的账本，就是 struct msqid_ds。它在 sys/msg.h 里定义。挑几个我们能用得上的字段讲透：
struct msqid_ds {
struct ipc_perm msg_perm; /* 上面的户口本，权限信息就嵌在里头 */
time_t msg_stime; /* 最近一次 msgsnd(2) 的时间 */
time_t msg_rtime; /* 最近一次 msgrcv(2) 的时间 */
time_t msg_ctime; /* 本结构最近一次被修改的时间 */
unsigned long __msg_cbytes; /* 当前队列里累计的字节数（非标准字段） */
msgqnum_t msg_qnum; /* 当前队列里的消息条数 */
msglen_t msg_qbytes; /* 这个队列最多允许的字节总数，就是 MSGMNB 的实际值 */
pid_t msg_lspid; /* 最近一次 msgsnd 的进程 PID */
pid_t msg_lrpid; /* 最近一次 msgrcv 的进程 PID */
};先提醒一个会命中的坑：你 include 的是 sys/msg.h 得到的这个用户态结构，字段名和我刚才引用的 linux/msg.h 内核头可能略有出入——比如内核头里那个"当前字节数"字段叫 __msg_cbytes（带下划线），而某些老版本 glibc 给用户态的字段命名又不完全一样。结论是：写代码时优先用 msg_qbytes、msg_qnum、msg_lspid、msg_lrpid 这些各家都稳定的名字，别死磕 __msg_cbytes。 我们在 msgctl 的 IPC_STAT 例程里就只用了稳定的那几个。
这几项里，msg_qbytes 是真正限制这个队列"实际能装多少"的上限，它是一份队列的字节余额；命中它时 msgsnd 就会阻塞。它不是预言中的死值，内核允许某些情况下用 IPC_SET 去调（往下调最容易，往上调超过 MSGMNB 需要特权）。我们到 msgsnd 和"上限"那一节会重点用它。
一句话搞懂 key、msqid 和内核对象的关系
很多初学者会被"key、msgid、内核对象"这三者搞晕。其实它们仨是一条清晰的链路：
- key：是队列的名字，全局字符串级的唯一标识，由两个毫无关系的进程约好同一个值就能"找到同一个队列"。名字不能凭空造，标准做法是用
ftok()从一个"存在的路径 + 一个整数子序号"算出。ftok是"file to key"的意思。 - msgid（msqid）：是进程本次打开的"门牌号"，由
msgget(key, ...)成功返回。它就像一个文件描述符，后续的msgsnd/msgrcv/msgctl全拿它做主语，而不再提着key到处跑。 - 内核对象：是真正储存在内核里的那一堆数据结构（
msqid_ds+ 消息链表）。key和msgid都只是指向它的"入口名"。
打个比方：内核对象是那间"仓库"，key 是仓库在系统里的注册名（比如"华东仓"），msgid 是你今天凭身份证领到的那把钥匙的编号。领了钥匙（msgget）之后，你进出仓库都只需报钥匙号，不用再背仓库名。
这里有个很容易踩的认知误区要打掉：在一个进程里，msgget 用了同一个 key 两次，会不会得到两个不同的队列？不会——它总是返回那"同一个队列"的 msgid。 只要 key 相同，就对应同一个内核对象。而 IPC_CREAT | IPC_EXCL 是为了保证"这队列必须是我新创建的，别人要是已经建了，就给我报 EEXIST"。这在"谁先创建、谁后打开"的多线程/多进程协同里特别常见——通常约定：服务方负责创建（带 IPC_EXCL 保证唯一），客户端负责打开（只带 IPC_CREAT 不挑不拣）。
ftok 的两个参数，第一个是已存在并且有读取权限的路径（常见的是 /tmp 这种谁都能读的目录），第二个是 0~255 的一个整数子序号。同一个路径 + 同一个子序号，在任何进程里算出结果都一样——这就是两个进程"约好名字"的方式。
#include <sys/types.h>
#include <sys/ipc.h>
#define PATHNAME "/tmp" /* ftok 用的路径，必须真实存在 */
#define PROJID 0x4321 /* 子序号，双方约定一致即可 */
key_t key = ftok(PATHNAME, PROJID); /* 得到唯一的 key */思考题（带详解答案）：为什么 ftok 的路径参数必须是磁盘上一个"真实存在"的文件，随便编一个"不存在的路径"会怎样？
答案：ftok 是通过"拷文件的实际属性"来生成 key 的——它把那个文件所在文件系统的 stat 里的设备号、inode 号这些信息，和子序号混合运算出一串整数。如果路径不存在，stat 拿不到任何文件属性，ftok 必然失败，返回 -1，errno 被置为 ENOENT（没有这个文件）。更麻烦的是，即便不走运"算出来"也一样，所以实际工程里用它时铁律是：先确认路径存在，再调 ftok，见到返回值是 -1 就必须处理错误。这个"路径不存在 → ftok 失败"是很多刚学消息队列的人第一次跑就莫名退出 exit(1) 的根源。
（顺带一提：ftok 生成的 key 其实规模有限、还可能"撞号"——现代工程里已有人主张用 key = 0x一串数字 的字面量起名，或者干脆用 IPC_PRIVATE 生成"无 key 的私有队列"。后面 msgctl 例程会用到 IPC_PRIVATE，它每次都保证开一个全新队列，谁也别想猜中它。）
msgget：一手创建、一手打开
接口签名和权限都先看全：
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
int msgget(key_t key, int msgflg);key：消息队列的名字，就是上面ftok算出来的那个（也可以写IPC_PRIVATE，表示"给我一个全新的、无 key 的队列"）。msgflg：由几部分组成——要么是权限位（0666之类），叠加下面这几种标志：IPC_CREAT：key 对应的队列不存在就创建它；IPC_EXCL：和IPC_CREAT一起用时，如果队列已经存在就报错EEXIST，保证"这个队列是我新建的"。
- 返回值：成功返回一个非负整数
msgid（队列标识码），失败返回-1并设置errno。
用法上有个我们反复强调的约定俗成，我再用代码写一次给你看——服务方创建，客户端打开：
/* 服务方：我要"凭空建立一个队列"，别人已经建了就报错 */
int serv_msqid = msgget(key, IPC_CREAT | IPC_EXCL | 0666);
if (serv_msqid < 0) { perror("msgget"); exit(1); } /* EEXIST 说明别人抢先建了 */
/* 客户端：我要"连接这个队列"，它已经存在就直接拿，不存在才给建 */
int cli_msqid = msgget(key, IPC_CREAT | 0666);
if (cli_msqid < 0) { perror("msgget"); exit(1); }为什么客户端不加 IPC_EXCL？因为这是个"谁先到谁建"的竞态：如果客户端在服务方建好之前就来了，而它又带了 IPC_EXCL，那它会因为队列还不存在而直接失败（ENOENT 之类）。实际项目里更稳妥的做法，是让负责建的进程先跑起来，客户端稍后再连。初学者最常见的翻车现场，就是把这两个进程顺序搞反，或者两边都带了 IPC_EXCL 导致其中一边必失败。记住结论：共享用"打开"（IPC_CREAT），创建专用"排他"（IPC_CREAT | IPC_EXCL）。
IPC_PRIVATE 是另一个有意思的用法：不管你传不传 key，它都保证返回一个崭新的、别人猜不到 id 的队列。它最适合的场景是父子进程 fork 之后通信——父进程先 msgget(IPC_PRIVATE, ...)，fork 出去的子进程天然继承了这个 msgid，两者不用对 key 做任何约定。唯一的代价是：用完必须手动 IPC_RMID，否则这个"无主"队列会一直占着内核资源。我们在 msgctl 例程里就用它。
msgsnd：把消息投进队列
发送端的签名：
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);逐参数拆：
msqid：msgget拿到的队列标识码。msgp：指向待发送消息的指针，即"struct msgbuf（或你自己定义的等价结构）的首地址"。它的第一个成员必须是long mtype，这是铁律，内核就是从这里读类型。msgsz：消息正文mtext的长度，单位是字节，而且绝不含那 8 个字节（64 位下）的long mtype。这是消息队列最经典的"长度单位"坑，见下。msgflg：控制"队列满了"时的行为。传0表示"队列满就阻塞等空位"；传IPC_NOWAIT表示"队列满立刻返回，报EAGAIN"，不阻塞。- 返回值：成功
0，失败-1。
这里必须把长度单位这个最容易错的点掰开揉碎。看这个结构：
struct MsgBuf {
long mtype; /* 8 字节（64 位） */
char mtext[1024]; /* 正文最多 1024 字节 */
};如果我用它发一句 "hello"：
struct MsgBuf msg;
msg.mtype = 1;
strcpy(msg.mtext, "hello"); /* 正文实际 5 字节 */
/* 正确：msgsz 传"正文实际字节数"，不含 mtype */
int n = msgsnd(msqid, &msg, 5, 0);如果把第二个长度误写成 sizeof(msg)（也就是 8+5），那么内核会把"类型那 8 个字节之后"的内容也当成正文发出去——虽然 mtype 因为结构体对齐通常正好是开头 8 字节，但一旦 mtext 是变长的、或者你发了二进制数据，这个"多几个尾巴字节"的错误会让你在接收端拿到一堆 \0 垃圾。绝对值规矩：msgsz 永远等于 strlen(正文)（对文本）或 实际有效字节数（对二进制），永远不等于 sizeof(struct MsgBuf)。 接收端的 msgrcv 同理。
阻塞规则也讲透：当队列里的累计字节数撞上 msg_qbytes（实际值 = MSGMNB）时，msgsnd 就"堵住"了——不传 IPC_NOWAIT 就死等，直到有别的进程取走了消息腾出空位；传了 IPC_NOWAIT 就立刻返回 EAGAIN，你可以自己决定去重试还是干别的。这条"满了就堵"是消息队列作为"有缓冲的通信"时的天然节流阀。
思考题（带详解答案）：msgsnd 长度单位写错（把 sizeof(struct MsgBuf) 当 msgsz）会导致什么现象？recv 端有何连带影响？
答案：第一，发送端会把 mtype 之后凡是 sizeof(struct MsgBuf) 那么长的字节统统当正文送出去，除了你真正要的那段文本，还捎带了结构体成员对齐产生的"空洞垃圾"和未初始化的尾字节。第二，接收端用 msgrcv 时如果缓冲区开得不够大、又没加 MSG_NOERROR，这"多加的尾巴"可能让接收长度超出缓冲区，从而触发 E2BIG 错误、消息还取不走。第三，更隐蔽的：如果你接收端拿 sizeof(...) 做 msgsz 去收、再用收到的长度去给字符串补 '\0'，多出的字节会让程序越界。结论：两端对 msgsz 的理解必须完全一致，统一按"正文实际字节数"走。 这背后正是因为消息队列"认字节数、不认字符串"，它干净、快速，但要求调用者心细。
msgrcv：按类型把消息取出来
接收端的签名：
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp, int msgflg);msqid：队列标识码。msgp：存放接收下来的消息的缓冲区首地址，结构照旧（long mtype+ 正文）。msgsz：缓冲区正文部分能容纳的最大字节数，同样不含mtype那 8 字节。msgtyp：决定了你要"点名取什么消息"，是本函数最灵魂的参数，四种规则见下一节。msgflg：控制"队里没有符合类型要求的消息"时的行为。传0就阻塞等待；传IPC_NOWAIT就立刻返回ENOMSG；还可叠加MSG_NOERROR（MSGSZ装不下就截断而不是报错）。- 返回值：成功返回实际拷贝进缓冲区正文的字节数（不含
mtype），失败返回-1。
跟 msgsnd 对应的阻塞规则在这里是：当队里没有满足 msgtyp 的消息时，msgrcv 会阻塞等待，直到有符合条件的消息被投进来。 用 IPC_NOWAIT 则立即以 ENOMSG 返回。你把它想成"在邮箱前站着等特定编号的信"：没有就等着，不想等了就走。
msgrcv 成功时，会在取消息时把对应消息移除出队列——它不像 msgctl 那样只读属性，而是真真切切把消息从内核链表里拿掉、拷进你的缓冲区。这一点和"文件读取要移动文件偏移"类似，别指望"读了一次还能再读第二次"。
msgtype 的四种取法与"优先级"模拟
msgtyp 是一个 long，它的四种取值规则值得做成表格，这是消息队列区别于管道的最独特之处：
msgtyp 取值 | 规则 |
|---|---|
msgtyp = 0 | 取队列第一条消息，不管类型是什么 |
msgtyp > 0 | 取队列第一条类型等于 msgtyp 的消息。如果没有，就跳过前面的等着 |
msgtyp < 0 | 取类型小于等于 |msgtyp| 的、且这些候选中类型值最小的那一条 |
msgtyp > 0 且 flag = MSG_EXCEPT | 取队列第一条类型不等于 msgtyp 的消息 |
规则 3 是很多人记不住的，我给它拆一层：msgtyp = -2 的意思不是"只取类型 2"，而是"在所有类型 ≤ 2 的消息里，挑类型号最小那条"。假如队列里有类型 3、1、5，你要 msgtyp = -2，那么类型 ≤2 的只有 1，于是取到类型 1。要是队列里是类型 2、1、4，msgtyp = -2 会在 {1, 2} 里挑最小的 1 取走（即使 2 排在队头）。
规则 3 顺带就是"用消息类型模拟优先级"的钥匙。工程里一个常见做法是约定：类型号越小越"急"、越优先，发送方把最紧急的消息用更小的 mtype 发，接收方循环里用 msgtyp = -1（或某个负值上限）来取"当前最优先的那条"。因为 msgtyp < 0 总是给你最小的合格类型，"最小类型"即"最高优先级"，这就天然是带优先级的队列。反过来你当然也可以约定"类型号越大优先级越高"，那就要用 msgtyp>0 搭配遍历去数，稍微麻烦点，所以编程里"越小越优先、用负数取号"是主流。
为了让你把这些规则"看进骨头里"，我写一个完整的、可 gcc 直接编译运行的演示程序，三个场景把 0 / >0 / <0 都走一遍。请把注释和 printf 里的取回结果对着看：
/* recv_demo.c —— 集中演示 msgrcv 按 msgtyp 取消息的三种规则
* 编译：gcc -o recv_demo recv_demo.c
* 运行：./recv_demo （要求 /tmp 已存在，可写） */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#define PATHNAME "/tmp" /* ftok 的路径，必须真实存在 */
#define PROJID 0x4321 /* 双方约定的子序号 */
#define SIZE 128 /* 消息正文缓冲区大小 */
/* 照抄内核认识的消息结构：第一个成员必须是 long 型 mtype */
typedef struct {
long mtype; /* 消息类型 */
char mtext[SIZE]; /* 消息正文 */
} MsgBuf;
/* 通用"送一条消息"的辅助函数：塞好类型和正文发出去 */
static void Put(int msqid, long type, const char *text)
{
MsgBuf m; /* 栈上开一块消息区 */
m.mtype = type; /* 写类型 */
strcpy(m.mtext, text); /* 写正文 */
/* 注意 msgsz 传 strlen(text) 而不是 sizeof 结构体 */
if (msgsnd(msqid, &m, (size_t)strlen(text), 0) < 0) {
perror("msgsnd"); /* 发送失败就报错退出 */
exit(1);
}
printf(" 发送 [类型%ld] %s\n", type, text);
}
/* msgtyp = 0：只取队列第一条，类型不论 */
static void GetAny(int msqid)
{
MsgBuf m; /* 接收缓冲区 */
ssize_t n = msgrcv(msqid, &m, SIZE, 0, 0); /* msgtyp=0 取队头 */
if (n < 0) { perror("msgrcv"); exit(1); } /* 失败退出 */
m.mtext[n] = '\0'; /* 消息队列认字节数，手工补一个结束符才好当字符串打印 */
printf(" 取到(类型==? 取队头) [类型%ld] %s\n", m.mtype, m.mtext);
}
/* msgtyp > 0：取队列第一条类型恰好等于 want 的消息 */
static void GetType(int msqid, long want)
{
MsgBuf m;
ssize_t n = msgrcv(msqid, &m, SIZE, want, 0); /* 只认类型 want */
if (n < 0) { perror("msgrcv"); exit(1); }
m.mtext[n] = '\0';
printf(" 取到(类型==%ld) [类型%ld] %s\n", want, m.mtype, m.mtext);
}
/* msgtyp < 0：在类型<=|bound|的候选中取类型最小那条 */
static void GetMin(int msqid, long bound) /* bound 传负数 */
{
MsgBuf m;
ssize_t n = msgrcv(msqid, &m, SIZE, bound, 0); /* msgtyp=bound(<0) */
if (n < 0) { perror("msgrcv"); exit(1); }
m.mtext[n] = '\0';
printf(" 取到(类型<=%ld 中最小的) [类型%ld] %s\n", bound, m.mtype, m.mtext);
}
int main(void)
{
key_t key = ftok(PATHNAME, PROJID); /* 生成 key */
if (key < 0) { perror("ftok"); exit(1); } /* 路径不存在会在这里挂掉 */
int msqid = msgget(key, IPC_CREAT | IPC_EXCL | 0666); /* 排他地建新队列 */
if (msqid < 0) { perror("msgget"); exit(1); } /* 已存在则 EEXIST */
/* ---- 场景一：msgtyp = 0，取队头第一条 ---- */
printf("== 场景一：msgtyp = 0 ==\n");
Put(msqid, 3, "type-3"); /* 先送类型3，它排队头 */
Put(msqid, 1, "type-1"); /* 再送类型1 */
Put(msqid, 2, "type-2"); /* 再送类型2，此时队列顺序 [3,1,2] */
GetAny(msqid); /* msgtyp=0 取走第一条，即类型3 */
/* ---- 场景二：msgtyp = 1，挑特定类型 ---- */
printf("== 场景二：msgtyp = 1 ==\n");
GetType(msqid, 1); /* 队里现在是 [1,2]，取到类型1 */
/* ---- 场景三：msgtyp = -2，取 <=2 中类型最小 ---- */
printf("== 场景三：msgtyp = -2 ==\n");
GetMin(msqid, -2); /* 队里现在是 [2]，<=2 的只有2，取到类型2 */
msgctl(msqid, IPC_RMID, NULL); /* 用完了必须删，否则残留内核 */
return 0;
}运行结果（去掉多余打印）应该是：
== 场景一：msgtyp = 0 ==
发送 [类型3] type-3
发送 [类型1] type-1
发送 [类型2] type-2
取到(类型==? 取队头) [类型3] type-3
== 场景二：msgtyp = 1 ==
取到(类型==1) [类型1] type-1
== 场景三：msgtyp = -2 ==
取到(类型<=-2 中最小的) [类型2] type-2思考题（带详解答案）：把场景三里 GetMin(msqid, -2) 改成 GetMin(msqid, 2)，会发生什么？
答案：GetMin(msqid, 2) 把 msgtyp 传成了正数 2，于是规则就变成了"取第一条类型恰好等于 2 的消息"。在场景三执行那一刻，队里唯一剩下的是类型 2 的消息，所以结果仍然是取到类型 2——好像"碰巧对了"。但一旦队列里同时存在类型 1 和类型 2，差异就出来了：msgtyp=2 会跳过类型 1、专挑类型 2；而 msgtyp=-2 会在 {1,2} 里挑最小的 1。这两个行为根本不同。这个例子就是要提醒你：正负号不是装饰，msgtyp 的正负直接改写取法语义，写代码时务必看清。 很多人想表达"取 <=2 最小的"却手滑写成正 2，拿 -2 才是"阈值式优先取小"，2 是"点名式精确取"。
msgctl：能删能查能改的"总管"
接口：
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
int msgctl(int msqid, int cmd, struct msqid_ds *buf);第三个参数 buf 在不同 cmd 下用途不同：IPC_STAT 时它作为"输出"接住内核数据；IPC_SET 时它作为"输入"把你要改的值送进去；IPC_RMID 时无所谓，传 NULL 即可。三个动作：
IPC_STAT：把内核里这个队列的msqid_ds拷一份到buf，让你能读到当前字节数、消息条数、最后发送/接收进程的 PID 等。这是排查"这个队列堵没堵、谁最后动过它"的利器。IPC_SET：用buf里的msg_perm（属主/权限）、msg_qbytes（能装总字节数上限）去覆盖内核里的属性。注意调整msg_qbytes有权限门坎：往小调一般不需要特权，想调到超过 MSGMNB 则需要 root。这也是"想故意把一个队列塞满来复现msgsnd阻塞"的常用手法：先用IPC_SET把msg_qbytes调小。IPC_RMID：删除队列。成功后这块队列立刻从内核消失，返回0；之后再有进程拿这个msgid去msgsnd/msgrcv，会得到EIDRM（队列已被删）错误。上面两个例程结尾都靠它收尾。
看一个用 IPC_STAT 读取属性的完整例程，顺带示范 IPC_PRIVATE 私有队列的一生：
/* msgctl_stat.c —— 用 IPC_STAT 读队列属性，并演示 IPC_PRIVATE 私有队列
* 编译：gcc -o msgctl_stat msgctl_stat.c
* 运行：./msgctl_stat */
#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
int main(void)
{
/* IPC_PRIVATE：每次都返回一个全新的、无 key 的队列 */
int msqid = msgget(IPC_PRIVATE, IPC_CREAT | 0600); /* 只有属主能用 */
if (msqid < 0) { perror("msgget"); exit(1); }
struct msqid_ds ds; /* 接收内核属性的缓冲区 */
/* IPC_STAT：把队列属性拷进 ds */
if (msgctl(msqid, IPC_STAT, &ds) < 0) {
perror("msgctl IPC_STAT"); exit(1);
}
printf("单队列字节上限 msg_qbytes = %lu\n", (unsigned long)ds.msg_qbytes);
printf("当前消息条数 msg_qnum = %lu\n", (unsigned long)ds.msg_qnum);
printf("上次 msgsnd 的 PID = %d\n", (int)ds.msg_lspid);
printf("上次 msgrcv 的 PID = %d\n", (int)ds.msg_lrpid);
/* 用完立刻 IPC_RMID，防止私有队列残留内核 */
if (msgctl(msqid, IPC_RMID, NULL) < 0) { perror("msgctl IPC_RMID"); exit(1); }
printf("队列已删除\n");
return 0;
}编译运行，你大概率会看到 msg_qbytes 打印出 16384（这正是 Linux 上 MSGMNB 的常见默认值）。如果没做任何收发，msg_qnum 是 0，两个 PID 都是 0（"上次"尚无）。
思考题（带详解答案）：用 IPC_STAT 读到 msg_qbytes 是 16384，我往队列里一条条发、每条正文刚好 1 字节，理论上能塞多少条 1 字节消息到"堵住"为止？
答案：msg_qbytes 管的是总字节数，不是总条数。每塞进一条消息，占用的字节除了正文那 1 字节，还要算上内核为消息追加的"信封"开销（8 字节 mtype + 若干内核链表元数据）。所以严格说，在 msgsnd 阻塞前能塞进去的条数并不等于 16384/1=1.6 万；真实空间由"每条消息的开销（约 8+16 左右的元数据字节，依内核版本而异）+ 正文"共同决定，条数通常明显少于你说的这个上界。这个例子想强调两件事：一是 msg_qbytes 是"字节"而非"条数"的单位，量化时别想当然；二是"队列能放多少"和"进程能不能写进去"之间隔着 msg_qbytes 这道实实在在的闸。 想精确观察就落到实验里去数，别纸上谈兵。
消息队列的生命周期：不清理真的会泄漏
这是本课最"反直觉"也最实际的一节。先立结论：消息队列的生命周期是"跟着内核走"的，进程退出、程序崩溃，都不影响它的存在。 也就是说，你写了一个程序 msgget 建了队列却忘了删，然后程序 return 0 退出了——那条队列还在，白白占着内核资源。只有两种方式能送走它：
- 程序里：
msgctl(msqid, IPC_RMID, NULL)，我们例程末尾都这么干； - 命令行：
ipcrm，这适合你"忘了写删除代码、队列已经残留"的救急场景。
命令行工具是 ipcs 和 ipcrm：
# 1. 查看系统里所有消息队列
ipcs -q
# 输出形如：
# ------ Message Queues --------
# key msqid owner perms used-bytes messages
# 0x00004321 2686976 @uu 666 6 1
# 2. 按 msqid（上表第二列的数字）删除某个队列
ipcrm -q 2686976
# 3. 也可以按 key 删除（key 要写成内核接受的数值形式）
ipcrm -Q 0x00004321平时自测写程序时，最省事也最不容易出错的两个动作是：进程内 IPC_RMID，以及排查残留时 ipcrm -q <msqid>（用第二列的 id，最直白）。ipcrm -Q <key> 因为要对 key 的进制度格式磕碰（十进制/十六进制），新手别首选它。
思考题（带详解答案）：我把程序运行了一遍、忘了写 IPC_RMID，进程退出了。请问这条消息队列现在是什么状态？我下一次再跑这个程序（同样的 key），会发生什么？
答案：进程退出后，队列仍然以"残留"状态躺在内核里，ipc -q 能看到它，占用其 used-bytes 与内核槽位。下一次再跑程序时，取决于你 msgflg 怎么写的：如果你只用 IPC_CREAT（不带 IPC_EXCL），msgget 会"重逢"这条旧队列，直接拿到它的 msgid——很可能队列里还躺着你上次没取走的消息，行为跟第一次跑（空队列）完全不同，容易造成"收到陈年旧数据"的错觉；如果你带了 IPC_CREAT | IPC_EXCL，这次会直接 EEXIST 失败，程序报错退出。所以消息队列的"残留"是实实在在影响后续运行的。养成"用完即删"的条件反射，是消息队列学习者最该带走的工作习惯之一。
动手写一个完整的通信小项目（C 语言）
理论铺完，我们来一个"服务端创建、客户端连接"的最小可运行组合。这个项目复刻了源材料里 MsgQueue.hpp 的客户端-服务端套路，但改写成了两个纯 C 文件，方便你一条 gcc 指令编译出来。
server.c——服务端，负责建队列、收一条消息、然后删队列：
/* server.c —— 消息队列服务端
* 编译：gcc -o server server.c
* 运行：先 ./server （它创建队列并阻塞等消息） */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#define PATHNAME "/tmp" /* ftok 的路径 */
#define PROJID 0x4321 /* 与客户端约定一致的子序号 */
#define SIZE 1024 /* 正文缓冲区大小 */
#define MSGTYPE 1 /* 约定：只接收类型为 1 的消息 */
/* 消息结构，第一个成员必须是 long mtype */
typedef struct {
long mtype; /* 消息类型 */
char mtext[SIZE]; /* 消息正文 */
} MsgBuf;
int main(void)
{
key_t key = ftok(PATHNAME, PROJID); /* 生成同一个 key */
if (key < 0) { perror("ftok"); exit(1); }
/* 服务端负责创建：IPC_CREAT | IPC_EXCL 保证队列必须是我新建的 */
int msqid = msgget(key, IPC_CREAT | IPC_EXCL | 0666);
if (msqid < 0) { perror("msgget"); exit(1); }
printf("服务端：队列创建完成 msqid=%d\n", msqid);
MsgBuf msg; /* 接收缓冲区 */
/* msgrcv 按类型 1 取：队里没有类型 1 就阻塞等待 */
ssize_t n = msgrcv(msqid, &msg, SIZE, MSGTYPE, 0);
if (n < 0) { perror("msgrcv"); exit(1); }
msg.mtext[n] = '\0'; /* 补结束符，当字符串打印 */
printf("服务端收到 [类型%ld]: %s\n", msg.mtype, msg.mtext);
/* 收完删队列，生命周期收干净 */
if (msgctl(msqid, IPC_RMID, NULL) < 0) { perror("msgctl"); exit(1); }
printf("服务端：队列已删除\n");
return 0;
}client.c——客户端，负责连接（不创建）、发一条消息就走：
/* client.c —— 消息队列客户端
* 编译：gcc -o client client.c
* 运行：服务端先启动，再 ./client */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#define PATHNAME "/tmp"
#define PROJID 0x4321
#define SIZE 1024
#define MSGTYPE 1 /* 发送类型也定成 1，服务端同款 */
typedef struct {
long mtype;
char mtext[SIZE];
} MsgBuf;
int main(void)
{
key_t key = ftok(PATHNAME, PROJID);
if (key < 0) { perror("ftok"); exit(1); }
/* 客户端只打开：IPC_CREAT 不加 IPC_EXCL，队列已在就直接拿来用 */
int msqid = msgget(key, IPC_CREAT | 0666);
if (msqid < 0) { perror("msgget"); exit(1); }
MsgBuf msg; /* 构造要发的消息 */
msg.mtype = MSGTYPE; /* 类型与约定一致 */
strcpy(msg.mtext, "hello from client");
/* msgsz 传"正文实际字节数"，不是 sizeof(MsgBuf) */
if (msgsnd(msqid, &msg, (size_t)strlen(msg.mtext), 0) < 0) {
perror("msgsnd"); exit(1);
}
printf("客户端：发送完成\n");
return 0;
}Makefile——一条 make 全编译（注意 clean 目标的下一行必须以 Tab 缩进，这是 Makefile 的语法硬规定）：
.PHONY: all clean # 声明伪目标：all 和 clean 不是文件
all: server client # 默认目标，依赖这俩可执行文件
server: server.c # server 依赖 server.c
# 下面这一行必须以 Tab 开头（不是空格），编译服务端
gcc -o server server.c
client: client.c # client 依赖 client.c
# 下面这一行同样必须以 Tab 开头，编译客户端
gcc -o client client.c
clean: # 清理目标
rm -f server client跑法演示：
make # 编译出 server client
./server & # 后台启动服务端（它阻塞等消息）
./client # 客户端发完即退
# 服务端打印：
# 服务端：队列创建完成 msqid=xxxx
# 服务端收到 [类型1]: hello from client
# 服务端：队列已删除思考题（带详解答案）：这个例子里客户端和服务端用了同一个 key、同一种消息类型 MSGTYPE=1。如果客户端发消息类型设为 2、而服务端 msgrcv 仍按类型 1 收，会发生什么？怎么改？
答案：如果两边类型不一致，服务端的 msgrcv(...,1,...) 会一直阻塞在等"类型为 1 的消息"，而队列里只有一个类型 2 的消息在候着，服务端永远取不到。客户端发完就退出了（队列里残留一条类型 2 的消息），服务端则干等——直到你 Ctrl+C 或者用别的进程把类型 2 取走。而且因为服务端在等消息、还没执行到 IPC_RMID，即使客户端退了，这条队列也还残留着。修复要么把客户端类型也改成 1（两边一致），要么让服务端按类型 2 收，否则消息"锁死"在队列里。 更深一层：这也是"消息类型是双方约定出来的协议"的体现——mtype 不是内核强制双方对齐的规矩，是你自己和组织人约定的 'API'，约定错了就得背锅。
上限到底是多少：MSGMAX / MSGMNB / MSGMNI
前面卖了不少关子，现在把这些"上限"一并摊开。System V 消息队列有三道限额（Linux 常见的默认值列在括号里，读者可通过 /proc/sys/kernel/msg* 在任意发行版上实测确认，随内核配置可能不同）：
| 宏 | 含义 | Linux 常见默认 |
|---|---|---|
MSGMAX | 单条消息的正文最大字节数 | 8192 |
MSGMNB | 单个队列最多能容纳的总字节数（也是 msqid_ds.msg_qbytes 的默认来源） | 16384 |
MSGMNI | 整个系统允许的消息队列总数上限（内核参数 msgmni） | 32000（新内核，可配） |
- 命中
MSGMAX：msgsnd时正文超过它，直接失败EINVAL，消息不发。 - 命中
MSGMNB（即msg_qbytes）：队列"满"了，msgsnd阻塞（或IPC_NOWAIT给EAGAIN），这就是我们反复讲的"满了就堵"。 - 命中
MSGMNI：再msgget建新队列会失败ENOSPC，"装不下更多队列"。
阶段性小结一条判断路径："这条消息能不能发"大部分时候由 MSGMAX 决定；"这个队列还能不能塞"由 MSGMNB 决定；"系统还能不能建队列"由 MSGMNI 决定。 三个数字平时记不大住没关系，真到"我怎么发不出去"时，按 EINVAL→查正文长度、EAGAIN/阻塞→查 msg_qbytes、ENOSPC→查 msgmni 三条线去对号入座即可。
（不确定处我在此注明：上面表格里 8192/16384/32000 是我依据主流 Linux 发行版内核默认值给的"常见值"，不代表所有环境。你的机器到底是多少，跑一句 cat /proc/sys/kernel/msgmax /proc/sys/kernel/msgmnb /proc/sys/kernel/msgmni 以实为准。）
责任链模式：把一串处理逻辑串成一条链
前面通信的部分，你已经能把消息从 A 送到 B 了。但真实的生产里，"收到一条消息"往往只是个开始——它后面总跟着一连串处理。我们沿用源材料里的那个经典需求升级，把它讲透：
新需求（之前只有一个"收发 hello"）：
client发给server的内容，要拼接上当前时间和进程 PID 信息；server收到的内容要持久化保存到文件里；- 如果文件内容过大，要切片保存，并在指定目录下打包成压缩包，命令自定义。
你会怎么组织"加工 → 落盘 → 备份"这三步？新手最容易写出的是一坨大函数：先拼时间戳，再写文件，再去数行数，超了再 fork+tar……三步逻辑全烩一锅，将来想"在某两步中间插个新步骤"、或者"临时关掉某一步"，都得回去动这一个函数，改一处冒一身汗。
责任链模式（Chain of Responsibility Pattern）就是来治这个病的。 它属于行为型设计模式。核心思想是俩：
- 把请求沿着"处理者链"传递：每个处理者先检查自己要不要处理这个请求，能处理就处理，处理完（或自己处理不了）就把请求交给链上的下一个处理者；
- 松耦合：请求的发送方不知道、也不必知道"到底有哪几环处理它"，它只对着链头说一句"干活"，具体的环节拆分、顺序、增减，全由责任链内部组织。这就把"发请求的人"和"具体收请求处理的人"解耦了。
再具象一点：责任链就像工厂里的流水线。你（发送方）把半成品扔到传输带起点（链头），流水线上每个工位（处理者）各自只干自己那一道工序，做完传给下一个工位，直到走完全线。哪个工位今天停工（enabled 关掉），你就把它闲置，流水线照样转，只是少了那道工序。
换成 C 语言，我们用"链表式处理器 + 函数指针"来落地这个模式——每个处理者是一个节点，节点里放三样东西：要不要干活（enabled）、下一环是谁（next）、以及它自己真正干活的函数（func）。一个统一入口 Dispatch 负责"处理当前节点 → 顺着 next 往下传"，直到链尾。整条链的顺序、增减节点，只需改 main 里怎么串，完全不用动各环节的实现。这就是解耦的甜头。
下面是一个完整的、可直接编译运行的 C 责任链实现，对应"格式化 → 存文件 → 超行数备份"三环：
/* chain.c —— 用 C 实现责任链：格式化 -> 存文件 -> 超行数打包备份
* 编译： gcc -o chain chain.c
* 运行： ./chain
* 说明： 会创建 ./tmp/ 目录并写 test.log，行数>MAX_LINES 时触发 tar 备份 */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <sys/stat.h>
#define INFO_SIZE 4096 /* info 文本缓冲区大小 */
/* ---------- 处理者节点：责任链中的一环 ---------- */
typedef struct Handler Handler; /* 前向声明，允许互相指 */
typedef void (*HandleFunc)(Handler *h, char *info);/* 节点真正的处理函数指针 */
struct Handler {
int enabled; /* 这个节点要不要干活（1 开 / 0 关） */
Handler *next; /* 链上的下一个节点，NULL 表示链尾 */
HandleFunc func; /* 本节点的处理函数 */
};
/* 统一入口：先处理当前节点（若 enabled），再把请求传给下一个节点 */
static void Dispatch(Handler *h, char *info)
{
if (h->enabled) /* 节点自己开着才执行它的工序 */
h->func(h, info); /* 调本节点的处理函数 */
if (h->next) /* 还有下一环吗？有就传下去 */
Dispatch(h->next, info);/* 递归地交给下一环继续加工 */
else
printf("责任链结束,处理完成\n"); /* 走到链尾，收个尾 */
}
/* ---------- 工序 1：格式化，给文本加上"时间 - PID -"前缀 ---------- */
static void Format(Handler *h, char *info)
{
char tmp[INFO_SIZE]; /* 构造新文本的临时缓冲 */
(void)h; /* 本工序用不到节点数据，抑制未用警告 */
printf("[Format] 加工中...\n");
/* 在原文前拼上时间戳和当前进程 PID，末尾加换行 */
snprintf(tmp, sizeof(tmp), "%ld - %d - %s\n",
(long)time(NULL), (int)getpid(), info);
strncpy(info, tmp, sizeof(tmp)); /* 把加工后的文本写回 info */
sleep(1); /* 模拟耗时，便于观察流水线节奏 */
}
/* ---------- 工序 2：保存，把文本追加写进 ./tmp/test.log ---------- */
static void Save(Handler *h, char *info)
{
FILE *fp; /* 文件指针 */
(void)h;
printf("[Save] 写入日志文件\n");
mkdir("./tmp", 0755); /* 确保目录存在 */
fp = fopen("./tmp/test.log", "a"); /* "a"=追加写，不清空旧内容 */
if (fp == NULL) { perror("fopen"); return; } /* 打开失败就算了 */
fputs(info, fp); /* 把格式化好的文本写进文件 */
fclose(fp); /* 关流，落盘 */
sleep(1);
}
/* ---------- 工序 3：备份，行数超过上限就把文件改名并打成 tar 包 ---------- */
#define MAX_LINES 5 /* 行数上限，设小便于快速触发备份 */
static void Backup(Handler *h, char *info)
{
int lines = 0; /* 统计 test.log 现有行数 */
int c; /* 逐字读文件用 */
FILE *fp; /* 文件指针 */
(void)h; (void)info;
printf("[Backup] 检查是否需要备份\n");
fp = fopen("./tmp/test.log", "r"); /* 打开日志统计行数 */
if (fp) {
while ((c = fgetc(fp)) != EOF) /* 逐字符读到文件尾 */
if (c == '\n') lines++; /* 数到换行就加一行 */
fclose(fp); /* 关流 */
}
printf("test.log 当前 %d 行（上限 %d）\n", lines, MAX_LINES);
if (lines > MAX_LINES) { /* 只有"超过"才备份 */
long ts = (long)time(NULL); /* 用时间戳做唯一文件名 */
char oldname[256], tarname[256]; /* 原始名与压缩包名 */
printf("行数超过上限,触发日志备份...\n");
/* 把 test.log 改名成 test.log.<时间戳>，腾出原目录让后续继续写新文件 */
snprintf(oldname, sizeof(oldname), "./tmp/test.log.%ld", ts);
if (rename("./tmp/test.log", oldname) != 0) { perror("rename"); return; }
snprintf(tarname, sizeof(tarname), "./tmp/test.log.%ld.tgz", ts);
pid_t pid = fork(); /* 开子进程去打包 */
if (pid == 0) { /* 子进程分支 */
/* exec 成 tar 命令：czf tarname oldname，注意要以 NULL 收尾 */
execlp("tar", "tar", "czf", tarname, oldname, (char *)NULL);
perror("execlp"); /* 只有 exec 失败才走到这 */
_exit(127); /* 子进程带着错误状态退出 */
}
waitpid(pid, NULL, 0); /* 父进程等子进程打包结束 */
remove(oldname); /* 删掉已打成包里的大体积原文件，只留 .tgz */
printf("备份完成,已生成 %s\n", tarname);
}
sleep(1);
}
int main(void)
{
/* 造三个责任链节点，初始都开启 */
Handler fmt = { 1, NULL, Format }; /* 格式化节点 */
Handler sav = { 1, NULL, Save }; /* 存文件节点 */
Handler bak = { 1, NULL, Backup }; /* 备份节点 */
/* 把三个节点串成一条链：fmt -> sav -> bak */
fmt.next = &sav; /* 格式化之后走存文件 */
sav.next = &bak; /* 存文件之后走备份 */
char info[INFO_SIZE]; /* 一条待处理的文本 */
for (int i = 1; i <= 7; i++) { /* 连扔 7 条，第 6 条会触发备份 */
snprintf(info, sizeof(info), "第%d条日志", i); /* 造一条消息内容 */
printf("\n---- 提交第 %d 条 ----\n", i);
Dispatch(&fmt, info); /* 对着链头只说一句:干活! */
}
return 0;
}这个链是怎么"转"起来的
Dispatch(&fmt, info) 就是那个"对着链头喊一声干活"的入口。它内部的逻辑是：如果当前节点开着（enabled），就执行它自己的 func；然后看有没有 next，有就递归继续，没有就打印"责任链结束"。所以整条链的执行方向是：Format 拼前缀 → Save 写入文件 → Backup 检查行数、超了就打包。而 enabled 字段给了你"临时关掉某道工序"的能力——你只要写成 fmt.enabled = 0，Dispatch 就会跳过格式化，直接把原始文本传给存文件的节点。这就是 EnableHandler 开关的美化版：加减工序、开关工序，都在装配阶段完成，不用动任何工序内部。 这也正是责任链最值钱的地方：每个处理者只关心自己这一环，以及背后的下一环；至于整条链怎么拼、要不要拼，那是装配者（使用者）的事。 新增一个处理环节，你只需再写一个带 func 的节点、在 main 里把它插到 next 之间即可，完全不影响既有环节的代码。
这个例子里有个刻意的小心机：MAX_LINES 定成 5、循环发放 7 条，所以第 6 条写入后行数到 6、超过 5，触发一次 rename + tar + remove；第 7 条写进的是备份后又新建的 test.log，行数只有 1，不再触发。你可以把 MAX_LINES 调大调小，亲眼看到"备份"这一环在链上"按需出现/消失"的效果。
责任链 vs 一串 if-else：区别在哪
你可能会嘀咕：这不就是把三个 if 串起来吗？差一层意思。if-else 是把"要不要做"的判定写死在某一处代码里，改一处逻辑要重编译整个函数；责任链把"每个环节做不做"（enabled）和"环节的执行顺序"（next 指针）设计成了可以运行时改的装配参数**。** 你要在运行时临时停用备份，if-else 得去改那段 if (lines > MAX_LINES) 的判断；责任链只要 bak.enabled = 0 一行。你要新增一道"加密"工序，if-else 得在函数体内硬插一段；责任链只要新增一个节点、在 main 里把它接到 Format 与 Save 之间。解耦的收益，规模越大越明显。
思考题（带详解答案）：把 Dispatch 的递归改成"节点函数内部自己转发 next"（即源材料里每个 Execute 结尾都调 _next_handler->Execute），两者等价吗？各有什么取舍？
答案：完全等价——这是责任链两种常见的落地写法。"统一入口转发"（本文做法）把"我做完之后立刻传给下一环"这个动作收进一个 Dispatch，各环节聚焦在自己的工序，代码更不容易写漏"忘传下一环"；但代价是链的"前进控制"集中在入口函数，若某个环节想"就地拦截、不再往后传"（比如校验失败的非法文本不想进文件），统一入口版要么得靠 enabled、要么得给节点加"终止"信号，稍绕。"节点内自发转发"（源材料做法）把转发动作和工序揉在一起，环节内部自由度大——想在特定条件下"吞掉请求不传下去"特别直接（只有它能决定自己传不传）；但代价是每写一个新环节都要记得在结尾调下一个，漏一次请求就悄悄断了链，找 bug 更难。工程上建议：若大多数环节都乖乖顺序执行，用"统一入口转发"；若环节里有大量"判一下要不要往下传"的分支，用"节点内自发转发"。 没有高下，取决于你的链"规不规矩"。
综合自测（每题都带详解答案）
一、判断对错并说明理由："消息队列里消息一定是先进先出取走的。"
详解：错。只有当 msgrcv 用 msgtyp = 0 时，才是严格按"队头先来先走"取的。一旦用了 msgtyp > 0（取指定类型），就可能跳过排队靠前的其他类型消息，先去取后面那条——它不是严格 FIFO，而是"FIFO 底子 + 类型挑选"混合体。规则 3（msgtyp < 0）还会在候选中挑"类型最小那条"，也不保证先来先走。
二、msgsnd 什么时候/在哪里会阻塞？ IPC_NOWAIT 又是帮了什么忙？
详解：当把新消息放进去会导致**队列累计字节数超过 msg_qbytes（即 MSGMNB 实际值）**时，msgsnd 暂停下来等——等有别的 msgrcv 把消息取走、腾出空位再继续。IPC_NOWAIT 的意思是"别等我，队列满了立刻回来告诉我"，此时返回 -1 且 errno=EAGAIN（可重试），让你不用干等着，可以先去处理别的逻辑。
三、两端对 msgsz 的约定不一致会造成什么？ 请给出一个具体例子。
详解：发送端 msgsnd 的 msgsz 是"本次要发的正文实际字节数"，接收端 msgrcv 的 msgsz 是"缓冲区能收的最大字节数"。若接收端的 msgsz 比某条实际消息的正文短：没加 MSG_NOERROR 就报 E2BIG 且消息不取走；加了 MSG_NOERROR 则截断到缓冲区大小。若发送端把 msgsz 错写成 sizeof(struct MsgBuf)（含 8 字节 mtype），会把结构体对齐空洞和多发的一截垃圾一起当正文发出去，接收端按长度拼接字符串可能得到乱码或越界。规范做法：两端都以"正文实际字节数"为单位，并保持一致。
四、我在 msgget 用了 IPC_CREAT | IPC_EXCL，结果报 EEXIST。这是 bug 吗？怎么处理？
详解：不一定是 bug。EEXIST 恰恰说明"这个 key 对应的队列已经存在"。业务上如果你就是要"独占创建"，说明另有进程（可能是上次残留没删、或别的服务抢先建了）已经建了这条队列；正确做法是识别场景：测试环境多半是上次忘了 IPC_RMID 的残留，用 ipcrm -q <msqid>（ipcs -q 查到）清掉再跑；生产环境则要判断是不是该"改建为打开"（改用纯 IPC_CREAT），或者对"谁负责创建"的职责做个约定，别让两个进程同时抢着建同 key 的队列。
五、为什么 ftok 对"路径是否真实存在"这么敏感？ 想确保两个无关进程拿到同一个 key，前提是什么？
详解：ftok 要用"该路径对应文件的设备号 + inode 号"这类 stat 信息去参与哈希生成 key。路径不存在 → stat 必然失败 → ftok 返回 -1。前提是：两个进程用同一个（真实存在且可 stat 的）路径 + 同一个 proj_id 子序号，因为 key = f(路径属性, proj_id)，两者任一不同算出来的 key 就不同，就会各自连到"不同的队列"，互相发消息却收不到。这也再次说明：key 的产生与传递是"双方约好的协议"，必须用代码层面能对得齐的常数（如共用的宏），而不是各写各的字面量。
六、"消息队列会阻塞"这件事，对并发编程既是优点也是风险，请各说一点。
详解：优点是它天然做了流量削峰/背压：生产者快、消费者慢时，队列会把消息攒在 msg_qbytes 允许的容量里，解放双方解耦；一旦满，msgsnd 阻塞，等于内核帮我们"踩刹车"，防止内存被无限撑爆。风险是死等/死锁：如果收发两端都阻塞着相互等待对方腾地方或投递某类型消息，可能谁也等不来，整体卡死；更现实的是，接收端用 msgtyp>0 等一条"永远不会来的类型"时，它会在那儿无限挂起（除非有超时/IPC_NOWAIT 策略）。实践中要设计好"超时 + 非阻塞 + 清理"，并时刻记住那句生命周期铁律——用完 IPC_RMID / ipcrm。
到这里，我们就把 System V 消息队列这条线从头捋到尾了。你认识了它"带类型、认字节数、随内核活"的三个灵魂，也亲手用四个函数——msgget 建队、msgctl 管队、msgsnd 投递、msgrcv 按类型收取——搭出了客户端-服务端的完整骨架。最难的那四类 msgtyp 取法我们用一张表和一段能跑的程序逐个敲实；MSGMAX/MSGMNB/MSGMNI 三道天花板，让你对"为什么塞不进、为什么建不出"心里有数；而"生命周期随内核、用完 ipcrm/IPC_RMID"这条，是绕不过也省不掉的清理功课。最后，我们从"收到消息之后怎么办"出发，用 C 实现的责任链，把"加工 → 落盘 → 备份"串成了一条解耦的流水线，并且看清了它和 if-else 的本质差别。
这一课的知识点很零碎、公式很多，但把它们串起来的那根线很简单：消息队列是内核里的"带类型邮箱"，它快、它有边界、它认类型，但你要自己管好清理、管好长度单位、管好阻塞与类型约定。 把这根线抓住，再把六个思考题自己推演一遍、把 recv_demo.c 和 chain.c 亲手敲过改过，你就算真正把这个 IPC 工具握在手里了。
下一步不妨自己动手试一试：把"责任链"那三环接到"消息队列的 msgrcv"后面——读出来的每一条消息都甩进链头 Dispatch，让格式化、落盘、备份成为消息的"后处理流水线"。这样一个"内核邮箱 + 处理链条"的小系统，会让你真切体会到 IPC 与设计模式配合起来有多顺。我们下次再见。
还没有评论 — 第一条由你来留。