035 System V 消息队列和信号量（了解）

小米里的大麦2025-07-162025-07-16

System V 消息队列和信号量（了解）

【Linux】进程间通信 4——system V 消息队列，信号量 | CSDN

System V IPC —- 消息队列详解 | CSDN 博客

消息队列的视频 & 博文 | YouTobe

System V 消息队列（编程接口指南）

信号量机制讲解

1. 消息队列的原理

System V 消息队列 是 UNIX/Linux 下的一种 进程间通信（IPC）机制，它允许不同进程以 消息（message）为单位交换数据，消息以 先进先出 的队列形式组织。异步通信：发送者发送后可以立即返回，接收者可随时读取。

System V 消息队列通过在内核中维护一个带有类型标记（mtype）的 FIFO 队列，实现了同一物理队列内多逻辑队列的分发能力，所有消息存在内核缓冲区，进程间通过 msgsnd/msgrcv 异步收发，OS 负责元数据管理与调度。

“如果只是一味地读数据，进程怎么知道哪些才是应该要读取的数据？”
这就是 mtype 的意义——把“一个大 FIFO”变成“多条逻辑子队列”。这也是 System V MQ 和单纯的管道（Pipe）的关键区别：管道只能 FIFO，消息队列支持 FIFO + 过滤 + 类型分发。

核心特征：

内核维护队列：由操作系统内核创建并管理 内存中的消息队列（不落盘，系统重启消失）。
消息结构
- 每条消息包含：
  - mtype：正整数类型标签（类型标识）。
  - mtext：数据块/正文数据，用户定义的内容（柔性数组）。
队列操作（先进先出）
- 发送/写操作 (msgsnd)：
  - 消息 追加到队尾（FIFO 基础）。
  - 队列满时默认阻塞，支持 IPC_NOWAIT 非阻塞。
- 接收/读操作 (msgrcv)：
  - 按 mtype 选择性读取：
    - type=0：读队首消息。
    - type>0：读匹配类型的 第一条消息（可打破 FIFO）。
    - type<0：读类型值 ≤ |type| 的最小类型消息。
  - 无匹配消息时默认阻塞。
生命周期与权限
- 独立于进程：队列需显式删除（msgctl(IPC_RMID)），否则持久存在。
- 通过权限结构（UID/GID、读写模式）控制访问。

当 OS 中存在多个消息队列，OS 自然也需要进行管理，这就又是 先描述再组织：先有数据的抽象描述（如 msgbuf），再有内核中的组织与管理（msqid_ds 结构和消息链表）。

2. 柔性数组

1. 什么是“柔性数组”？

柔性数组 简称 FAM，是 C99 引入的一种特殊语法：结构体的最后一个成员可以声明为大小未知的数组，用来在结构体后面动态附加可变长度的数据。作用： 允许在一个结构体后面连续存放额外的数据，而不需要在结构体里写死数组长度。

2. System V 消息队列里的体现

来看标准定义（man 手册示例）：

struct msgbuf
{
    long mtype;    // 消息类型
    char mtext[1]; // 消息正文
};

这里的 mtext[1] 很多人看到会觉得奇怪：为什么写 [1] 不是写 [0]？ 这是典型的 柔性数组的老写法，在老式 UNIX C 库中常这么用，因为 C90 没有正式的 [0] 或 [] 语法。在现代写法里，更标准的是：

struct msgbuf
{
    long mtype;
    char mtext[]; // 正确的 C99 柔性数组
};

3. 发送时是怎么用的？

当调用 msgsnd 时，内核不会管你 mtext 写多长，它只看你 msgsz 参数里指定的大小。
所以 mtext 只是占个位，真正的内存是由你自己分配：
1
struct mymsg* m = malloc(sizeof(long) + real_size);

这就是柔性数组的经典用法：结构体头 + 动态数据块，组成真正的消息。System V 消息队列的 msgbuf：是一个 固定头（mtype）+ 可变长正文（mtext） 的数据结构。mtext 本质是柔性数组（FAM）。这让同一个消息结构既可以描述元信息，也可以承载变长数据。传给内核时，内核只看 msgsz，并按字节拷贝。System V 消息队列用 FAM 来封装任意长度的消息内容，保证 msgbuf 在同一个内存块中连续存储 mtype 和 mtext，简化了内核拷贝和用户态对齐。

3. 系统调用（类比共享内存中 `shmget` 等函数）

1. `ftok` —— 生成唯一的 `key_t`，供 `msgget` 使用

上一节已经讲过 ftok，使用方法一模一样就不多赘述了，对于下面的 msgget 同样适用，同样的 非必须但推荐！

1
2
3

#include <sys/types.h>
#include <sys/ipc.h>
key_t ftok(const char *pathname, int proj_id);

2. `msgget` —— 创建

1. 功能

创建或打开一个消息队列。

2. 函数原型

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
int msgget(key_t key, int msgflg);

3. `msgflg` 常用标志

IPC_CREAT：若不存在则创建。
IPC_EXCL：和 IPC_CREAT 一起用，若已存在则报错。
权限位如 0666（所有用户可读写），0644（所有者读写，其他用户只读）。

4. 返回值

成功：消息队列标识符（非负整数）。
失败：-1，设置 errno。常见错误：
- EEXIST：IPC_CREAT|IPC_EXCL 时队列已存在。
- ENOENT：队列不存在且未指定 IPC_CREAT。
- EACCES：权限不足。

3. `msgsnd` —— 发送/写操作

1. 功能

发送消息。

2. 函数原型

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);

3. 参数

msqid：队列 ID。
msgp：消息结构体指针。
msgsz：不包括 mtype 的消息长度。
msgflg：可用 IPC_NOWAIT 非阻塞。

4. 返回值

成功：0
失败：-1，设置 errno。常见错误：
- EAGAIN：非阻塞模式下队列满。
- EIDRM：队列被删除。
- EACCES：无写权限。
- EINVAL：参数无效。

4. `msgrcv` —— 接收/读操作

1. 功能

接收消息。

2. 函数原型

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long mtype, int msgflg);

3. 参数详解

参数	含义	关键点
`msqid`	消息队列 ID	`msgget` 得到
`msgp`	接收消息的结构体指针	`msgbuf*`
`msgsz`	正文大小（不含 mtype）	`sizeof(mtext)`
`mtype`	要接收的类型	`0` 表示任意类型
`msgflg`	标志	`0` 或 `IPC_NOWAIT`

4. 返回值

成功：实际接收的字节数（mtext 长度）
失败：-1，设置 errno。常见错误：
- ENOMSG：无匹配消息（非阻塞模式）。
- E2BIG：消息过大且未指定 MSG_NOERROR。
- EIDRM：队列被删除。

5. `msgctl` —— 删除（控制）

1. 功能

对队列做控制操作。常用作删除操作。

2. 函数原型

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
int msgctl(int msqid, int cmd, struct msqid_ds *buf);

3. 参数详解

常用 cmd：

IPC_RMID：删除消息队列（最常用）。
IPC_STAT：获取队列状态。
IPC_SET：设置参数。

4. 返回值

成功：0。
失败：-1，设置 errno。常见错误：
- EIDRM：队列已被删除。
- EPERM：权限不足。
- EINVAL：无效队列 ID。

如何正确删除？

消息队列是内核资源，如果 不显式删除：

进程结束后也不会自动销毁
容易出现“僵尸消息队列”，占用系统 IPC 表

1	msgctl(msqid, IPC_RMID, NULL); // 删除队列正确做法

查看所有消息队列：
1
>ipcs -q
删除指定 ID 的消息队列：
1
>ipcrm -q 65536  # 替换为实际的 msqid

4. 代码示例

1. comm.hpp 文件

#pragma once
#include <iostream>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <cstring>
#include <stdlib.h>
using namespace std;

struct my_msgbuf
{
    long mtype;
    char mtext[1024];
};

const int key = ftok("./comm.hpp", 66);

2. consumer.cpp 文件

#include "comm.hpp"

int main()
{
    int key_t = key;
    int msgid = msgget(key_t, IPC_CREAT | 0666);
    if(msgid < 0)
    {
        perror("msgget failed");
        exit(1);
    }

    my_msgbuf msg;
    msgrcv(msgid, &msg, sizeof(msg), 1, 0);
    cout << "消息类型：" << msg.mtype << endl;
    cout << "消息内容：" << msg.mtext << endl;

    msgctl(msgid, IPC_RMID, NULL);
    cout << "队列已删除" << endl;

    return 0;
}

3. producer.cpp 文件

#include "comm.hpp"

int main()
{
    int key_t = key;
    int msgid = msgget(key_t, IPC_CREAT | 0666);
    if(msgid < 0)
    {
        perror("msgget failed");
        exit(1);
    }

    my_msgbuf msg;
    msg.mtype = 1;
    strcpy(msg.mtext, "Hello, linux!");

    if(msgsnd(msgid, &msg, sizeof(msg), IPC_NOWAIT) < 0)
    {
        perror("msgsnd failed");
        exit(1);
    }
    
    cout << "消息发送成功！" << endl;

    return 0;
}

4. 运行示例

5. 信号量（线程详解）

1. 背景

在并发环境下（多进程、多线程），多个执行流共享同一份资源（如共享内存、文件、缓冲区）时，若没有同步机制保护，就可能出现 脏读、脏写 等现象，导致 数据不一致。

例如：

进程 A 正在往一块共享内存写数据，还没写完；
进程 B 此时读了这块内存，读到的是 未写完整 的半成品数据；
导致 B 拿到的结果是错误的。

这就是典型的 并发访问导致的数据不一致。

2. 解决方案：互斥访问

为了保证数据一致性，需要在多个执行流之间做互斥，即：

任何时刻 同一份共享资源 只能被 一个执行流访问。
这里的“访问”是指执行对资源有读/写影响的那段代码。

3. 临界资源 & 临界区

临界资源： 指在同一时间内，只允许一个执行流访问的资源，比如：

共享内存
公共变量
公共缓冲区
I/O 设备等

临界区： 指的是 访问临界资源的那段代码，也就是：

整个程序中，可能只有几行到几十行代码真正操作共享资源；
只要保证这段代码是 互斥执行 的，数据就一致！

4. 如何实现互斥？

最常用的是加锁，锁的作用：只允许一个执行流进入临界区，其它执行流需要等待锁释放。锁的概念暂时还没学，放到后面讲。

5. 小结

概念	含义
共享资源	被多个执行流共享，且需要保护的资源
临界资源	需要互斥访问的共享资源
临界区	访问临界资源的那段代码
互斥	保证同一时刻仅一个执行流访问临界区
加锁	实现互斥的一种技术手段

6. 理解信号量

信号量在旧的书籍中也称信号灯，其本质是一把计数器，类似 int count = n（类似不代表等于！）。

它的值代表了 可用临界资源 的数量。
应用程序通过 申请（P 操作） 和 释放（V 操作） 这个计数器来间接地申请和释放临界资源。

申请（P 操作） 和 释放（V 操作） 是 原子的！原子的意思是：要么不做，要做就做完 —— 两态的，没有“正在做”这样的概念。

生活案例理解信号量和临界资源： 临界资源里面其实被划分很多小块，像一个电影厅 100 个座位，它最多就只能卖 100 张票，当我们还没去看电影先买票的时候，买票的本质就是对资源的预定机制，每卖出一张票，计数器就减减，放映厅里面的资源就减少一个，当计数器减到 0，资源已经被申请完毕了。

信号量模型	电影院模型
计数器初始值 = 可用资源数	初始票数 = 放映厅座位总数（比如 100 张票）
P 操作（wait）→ 计数器–1	买走一张票 → 余票数量 –1；若余票 > 0，则买票成功；若余票 = 0，则排队（阻塞）
V 操作（signal）→ 计数器+1	退票 → 余票 +1；若有人在排队，就给他一张票（唤醒）
计数器抵达 0	余票卖完，后续的买票请求都会被放到队列里，直到有人退票

如果电影院放映厅里面只有一个座位呢（超级 vip）？我们只需要一个值为 1 的计数器，只有一个人能抢到，只有一个人能进放映厅看电影，看电影期间只有一个执行流在访问临界资源，这就是互斥，我们把值只能为 1，0 两态的计数器叫做 二元信号量，本质就是一个锁（互斥锁）。

只有一个买票（P）能成功，其它排队等待；
相当于把“100 个座位”合并成“1 个整体座位”，要么全占，要么全空。

7. System V 信号量系统调用和使用流程（最难，多线程部分详解）

步骤	系统调用	作用
创建/获取	`semget()`	创建或获取一个信号量集
初始化	`semctl()`	设置初始值（如资源总量）
P/V 操作	`semop()`	对信号量执行加减操作
销毁	`semctl()`	删除信号量集（资源回收）

8. 信号量为什么属于 IPC？

进程间通信（IPC） 的广义定义是：“任何让两个或多个独立进程之间交换信息或状态 的机制，都是 IPC。”

信号量（System V）本质上是 由内核维护的一个计数器集合，这块计数器是：

存在内核内存里（和文件描述符一样，有唯一的 semid）。
可以被 多个进程同时打开、读写。
状态变更（P/V）会直接影响其他进程是否能继续往下执行。

所以信号量不是像管道那样直接“传递业务数据”，而是“传递同步信息”：“我这边可以/不可以用了！”这种状态的传递本身就是通信。

理解核心：

通信不等于一定要传业务数据，同步信息本身就是信息。
- 例：A 说“我写好了”，B 听见了就去读 → 这就是一种“信号通信”。
信号量是 所有需要同步的进程都能看到的：
- 通过 semget() 创建信号量集，返回的 semid 类似于管道文件描述符，多个进程只要 ftok + semget 相同，就能访问同一个信号量集。
- 信号量在内核里是全局可见，能跨父子进程，甚至无亲缘关系的进程（只要 key 一样，权限对得上）。

所以：通信不仅是传数据，互相协同也是通信。要协同，就必须有一块能被所有协同方看见、且可修改的共享状态，这就是信号量。

System V 消息队列和信号量（了解）

1. 消息队列的原理

2. 柔性数组

1. 什么是“柔性数组”？

2. System V 消息队列里的体现

3. 发送时是怎么用的？

3. 系统调用（类比共享内存中 shmget 等函数）

1. ftok —— 生成唯一的 key_t，供 msgget 使用

2. msgget —— 创建

1. 功能

2. 函数原型

3. msgflg 常用标志

4. 返回值

3. msgsnd —— 发送/写操作

1. 功能

2. 函数原型

3. 参数

4. 返回值

4. msgrcv —— 接收/读操作

1. 功能

2. 函数原型

3. 参数详解

4. 返回值

5. msgctl —— 删除（控制）

1. 功能

2. 函数原型

3. 参数详解

4. 返回值

4. 代码示例

1. comm.hpp 文件

2. consumer.cpp 文件

3. producer.cpp 文件

4. 运行示例

5. 信号量（线程详解）

1. 背景

2. 解决方案：互斥访问

3. 临界资源 & 临界区

4. 如何实现互斥？

5. 小结

6. 理解信号量

7. System V 信号量系统调用和使用流程（最难，多线程部分详解）

8. 信号量为什么属于 IPC？

小米里的大麦

3. 系统调用（类比共享内存中 `shmget` 等函数）

1. `ftok` —— 生成唯一的 `key_t`，供 `msgget` 使用

2. `msgget` —— 创建

3. `msgflg` 常用标志

3. `msgsnd` —— 发送/写操作

4. `msgrcv` —— 接收/读操作

5. `msgctl` —— 删除（控制）