Linux System V IPC 进阶：消息队列、信号量与内核管理解析

1.2.2 核心 API 使用

（1）ftok：生成唯一 Key（与共享内存通用）

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

• 作用：将'文件路径 + 项目 ID'转换为唯一 key，作为消息队列的全局标识；
• 注意：路径必须是存在的文件，proj_id 为非 0 整数（如 0x6666）。

（2）msgget：创建 / 获取消息队列

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

• 参数：
  • key：ftok 生成的 Key；
  • msgflg：权限标志，常用组合：
    • IPC_CREAT：不存在则创建，存在则获取；
    • IPC_CREAT | IPC_EXCL：不存在则创建，存在则报错；
  • 权限位（如 0666）：控制进程对队列的访问权限；
• 返回值：成功返回消息队列 ID（msgid），失败返回 -1。

（3）msgsnd：发送消息

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

• 参数：
  • msqid：msgget 返回的消息队列 ID；
  • msgp：指向消息结构体的指针（需自定义，首字段必须是 long mtype 消息类型）；
  • msgsz：消息正文长度（不含消息类型字段）；
  • msgflg：发送标志（0 阻塞，IPC_NOWAIT 非阻塞）；
• 返回值：成功返回 0，失败返回 -1。

（4）msgrcv：接收消息

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

• 参数：
  • mtype：接收消息的类型，0 表示接收第一个消息，>0 表示接收指定类型，<0 表示接收小于该值的所有类型；
  • 其他参数同 msgsnd。

（5）msgctl：控制消息队列（核心功能：删除）

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

• 参数：
  • cmd：控制命令（核心为 IPC_RMID，删除消息队列）；
  • buf：存储队列属性的结构体指针（IPC_RMID 时可设为 NULL）；
• 返回值：成功返回 0，失败返回 -1。

1.3 实战案例：消息队列实现 C/S 通信

通过'服务端 + 客户端'演示消息队列的使用，服务端接收客户端消息并回复，客户端发送消息并接收回复。

1.3.1 公共头文件（comm.h）

#ifndef COMM_H_
#define COMM_H_
#include <stdio.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <string.h>

#define PATHNAME "."
#define PROJ_ID 0x6666
#define MSG_SIZE 1024

// 消息结构体（首字段必须是 long mtype）
typedef struct msgbuf {
    long mtype; // 消息类型（服务端→客户端：100，客户端→服务端：200）
    char mtext[MSG_SIZE]; // 消息正文
} msg_t;

// 创建消息队列
int createMsgQueue();
// 获取消息队列
int getMsgQueue();
// 发送消息
int sendMsg(int msgid, long type, const char *text);
// 接收消息
int recvMsg(int msgid, long type, char *text);
// 删除消息队列
int destroyMsgQueue(int msgid);

#endif

1.3.2 公共实现（comm.c）

#include "comm.h"

// 内部通用函数：创建/获取消息队列
static int commMsgQueue(int flags) {
    key_t key = ftok(PATHNAME, PROJ_ID);
    if (key < 0) {
        perror("ftok error");
        return -1;
    }
    int msgid = msgget(key, flags);
    if (msgid < 0) {
        perror("msgget error");
        return -2;
    }
    return msgid;
}

// 创建全新消息队列
int createMsgQueue() {
    return commMsgQueue(IPC_CREAT | IPC_EXCL | 0666);
}

// 获取已存在的消息队列
int getMsgQueue() {
    return commMsgQueue(IPC_CREAT);
}

// 发送消息
int sendMsg(int msgid, long type, const char *text) {
    msg_t msg;
    msg.mtype = type;
    strncpy(msg.mtext, text, MSG_SIZE - 1);
    if (msgsnd(msgid, &msg, strlen(msg.mtext), 0) < 0) {
        perror("msgsnd error");
        return -1;
    }
    return 0;
}

// 接收消息
int recvMsg(int msgid, long type, char *text) {
    msg_t msg;
    ssize_t n = msgrcv(msgid, &msg, MSG_SIZE, type, 0);
    if (n < 0) {
        perror("msgrcv error");
        return -1;
    }
    msg.mtext[n] = '\0';
    strcpy(text, msg.mtext);
    return 0;
}

// 删除消息队列
int destroyMsgQueue(int msgid) {
    if (msgctl(msgid, IPC_RMID, NULL) < 0) {
        perror("msgctl error");
        return -1;
    }
    return 0;
}

1.3.3 服务端（server.c）

#include "comm.h"
#include <unistd.h>

int main() {
    // 1. 创建消息队列（通信发起者）
    int msgid = createMsgQueue();
    if (msgid < 0) {
        return 1;
    }
    printf("服务端：创建消息队列成功，msgid=%d\n", msgid);

    // 2. 循环接收客户端消息（类型 200）并回复（类型 100）
    char buf[MSG_SIZE];
    while (1) {
        memset(buf, 0, sizeof(buf));
        // 接收客户端消息
        if (recvMsg(msgid, 200, buf) < 0) {
            break;
        }
        printf("服务端收到：%s\n", buf);
        // 回复客户端
        if (strcmp(buf, "quit") == 0) {
            sendMsg(msgid, 100, "客户端退出，服务端即将关闭");
            break;
        }
        sendMsg(msgid, 100, "已收到你的消息");
        sleep(1);
    }

    // 3. 删除消息队列
    destroyMsgQueue(msgid);
    printf("服务端：消息队列已删除\n");
    return 0;
}

1.3.4 客户端（client.c）

#include "comm.h"
#include <unistd.h>

int main() {
    // 1. 获取消息队列
    int msgid = getMsgQueue();
    if (msgid < 0) {
        return 1;
    }
    printf("客户端：获取消息队列成功，msgid=%d\n", msgid);

    // 2. 循环发送消息（类型 200）并接收回复（类型 100）
    char buf[MSG_SIZE];
    while (1) {
        memset(buf, 0, sizeof(buf));
        printf("客户端请输入：");
        fflush(stdout);
        fgets(buf, MSG_SIZE, stdin);
        buf[strlen(buf) - 1] = '\0'; // 去除换行符

        // 发送消息
        sendMsg(msgid, 200, buf);
        if (strcmp(buf, "quit") == 0) {
            break;
        }

        // 接收回复
        memset(buf, 0, sizeof(buf));
        recvMsg(msgid, 100, buf);
        printf("服务端回复：%s\n", buf);
    }
    printf("客户端：退出通信\n");
    return 0;
}

1.3.5 编译与运行

# 编译
gcc server.c comm.c -o server
gcc client.c comm.c -o client

# 终端 1：启动服务端
./server
# 输出：服务端：创建消息队列成功，msgid=123456

# 终端 2：启动客户端
./client
# 输出：客户端：获取消息队列成功，msgid=123456
# 客户端请输入：hello
# 服务端回复：已收到你的消息
# 客户端请输入：quit
# 服务端：收到：quit
# 服务端：消息队列已删除

1.4 消息队列避坑指南

• 消息结构体必须以 long mtype 开头：这是内核规定的格式，否则 msgsnd/msgrcv 会报错；
• 消息类型必须为正整数：mtype 不能为 0 或负数，否则发送失败；
• 消息长度不含 mtype 字段：msgsnd 的 msgsz 参数是消息正文长度，不是整个结构体长度；
• 必须手动删除队列：消息队列不会随进程退出而释放，需用 msgctl(IPC_RMID) 删除，残留队列可通过 ipcs -q 查看、ipcrm -q msgid 删除。

二、System V 信号量：同步与互斥的核心工具

信号量并非用于数据传输，而是用于保护临界资源，实现进程间的同步与互斥。它本质是一个'内核维护的计数器'，通过 P/V 操作（申请/释放资源）控制进程对临界资源的访问。

2.1 核心概念铺垫

在学习信号量前，需明确三个关键概念：

• 临界资源：多个进程共享的资源（如共享内存、文件），一次仅允许一个进程访问；
• 临界区：进程中访问临界资源的代码段；
• 同步与互斥：
  • 互斥：任意时刻仅允许一个进程进入临界区（如多个进程写共享内存）；
  • 同步：多个进程访问临界资源时需遵循特定顺序（如'生产者先写，消费者再读'）。

信号量的核心是'通过计数器控制资源访问权限'，计数器值代表'可用资源数量'：

• P 操作（申请资源）：计数器 - 1，若计数器 < 0，进程阻塞；
• V 操作（释放资源）：计数器 + 1，若计数器≤0，唤醒一个阻塞进程。

2.2 核心原理与特性

2.2.1 底层实现逻辑

System V 信号量是一个'信号量集'（可包含多个信号量），内核通过 struct sem_array 管理信号量集属性，每个信号量通过 struct sem 描述：

struct sem {
    int semval;           // 信号量计数器值
    pid_t sempid;         // 最后一次操作该信号量的进程 PID
    unsigned short semncnt; // 等待计数器>0 的进程数
    unsigned short semzcnt; // 等待计数器=0 的进程数
};

2.2.2 核心特性

• 本质是计数器：不存储数据，仅用于权限控制；
• 生命周期随内核：信号量集创建后，需手动删除，否则残留于内核；
• 原子操作：P/V 操作是原子的，避免并发冲突；
• 支持多资源控制：一个信号量集可包含多个信号量，分别控制不同临界资源。

2.3 核心 API 详解

System V 信号量的 API 与消息队列、共享内存风格一致，核心包括 semget、semop、semctl。

2.3.1 semget：创建 / 获取信号量集

#include <sys/sem.h>
int semget(key_t key, int nsems, int semflg);

• 参数：
  • nsems：信号量集中的信号量个数（创建时必须指定，获取时可设为 0）；
  • 其他参数与 msgget 一致；
• 返回值：成功返回信号量集 ID（semid），失败返回 -1。

2.3.2 semop：执行 P/V 操作（核心 API）

#include <sys/sem.h>
int semop(int semid, struct sembuf *sops, size_t nsops);

• 参数：
  • sops：指向 struct sembuf 数组的指针，每个元素描述一个信号量的操作；
  • nsops：sops 数组的长度（操作的信号量个数）；
• struct sembuf 结构体：

struct sembuf {
    unsigned short sem_num; // 信号量集中的信号量下标（从 0 开始）
    short sem_op;           // 操作类型（-1=P 操作，+1=V 操作）
    short sem_flg;          // 操作标志（0=阻塞，IPC_NOWAIT=非阻塞）
};

2.3.3 semctl：控制信号量集

#include <sys/sem.h>
int semctl(int semid, int semnum, int cmd, ...);

• 参数：
  • semnum：信号量集中的信号量下标（操作单个信号量时指定）；
  • cmd：控制命令（核心命令如下）；
  • 可变参数：根据 cmd 不同，可传入 union semun 结构体（用于设置信号量初始值）；

核心命令说明：

2.3.4 union semun 结构体（需手动定义）

union semun {
    int val;                // 用于 SETVAL/GETVAL
    struct semid_ds *buf;   // 用于 IPC_STAT/IPC_SET
    unsigned short *array;  // 用于 SETALL/GETALL
    struct seminfo *__buf;  // 用于 IPC_INFO
};

2.4 实战案例：信号量保护共享内存

结合共享内存和信号量，实现'多进程安全写共享内存'——通过信号量的互斥机制，确保同一时刻仅一个进程写入共享内存，避免数据混乱。

2.4.1 公共头文件（sem_comm.h）

#ifndef SEM_COMM_H_
#define SEM_COMM_H_
#include <stdio.h>
#include <sys/ipc.h>
#include <sys/sem.h>
#include <sys/shm.h>
#include <string.h>

#define PATHNAME "."
#define PROJ_ID 0x6666
#define SHM_SIZE 1024
#define SEM_NUM 1 // 信号量集中的信号量个数

// 定义 semun 联合体
union semun {
    int val;
    struct semid_ds *buf;
    unsigned short *array;
    struct seminfo *__buf;
};

// 创建信号量集
int createSemSet(int nsems);
// 获取信号量集
int getSemSet(int nsems);
// 初始化信号量值
int initSem(int semid, int semnum, int val);
// P 操作（申请资源）
int P(int semid, int semnum);
// V 操作（释放资源）
int V(int semid, int semnum);
// 删除信号量集
int destroySemSet(int semid);

// 共享内存相关函数（复用之前的逻辑）
int createShm(int size);
int getShm(int size);
void *attachShm(int shmid);
int detachShm(void *addr);
int destroyShm(int shmid);

#endif

2.4.2 公共实现（sem_comm.c）

#include "sem_comm.h"

// 信号量集通用函数
static int commSemSet(int nsems, int flags) {
    key_t key = ftok(PATHNAME, PROJ_ID);
    if (key < 0) {
        perror("ftok error");
        return -1;
    }
    int semid = semget(key, nsems, flags);
    if (semid < 0) {
        perror("semget error");
        return -2;
    }
    return semid;
}

// 创建信号量集
int createSemSet(int nsems) {
    return commSemSet(nsems, IPC_CREAT | IPC_EXCL | 0666);
}

// 获取信号量集
int getSemSet(int nsems) {
    return commSemSet(nsems, IPC_CREAT);
}

// 初始化信号量
int initSem(int semid, int semnum, int val) {
    union semun un;
    un.val = val;
    if (semctl(semid, semnum, SETVAL, un) < 0) {
        perror("semctl SETVAL error");
        return -1;
    }
    return 0;
}

// P 操作
int P(int semid, int semnum) {
    struct sembuf sb;
    sb.sem_num = semnum;
    sb.sem_op = -1; // P 操作：计数器 -1
    sb.sem_flg = 0; // 阻塞模式
    if (semop(semid, &sb, 1) < 0) {
        perror("semop P error");
        return -1;
    }
    return 0;
}

// V 操作
int V(int semid, int semnum) {
    struct sembuf sb;
    sb.sem_num = semnum;
    sb.sem_op = 1; // V 操作：计数器 +1
    sb.sem_flg = 0;
    if (semop(semid, &sb, 1) < 0) {
        perror("semop V error");
        return -1;
    }
    return 0;
}

// 删除信号量集
int destroySemSet(int semid) {
    if (semctl(semid, 0, IPC_RMID) < 0) {
        perror("semctl IPC_RMID error");
        return -1;
    }
    return 0;
}

// 共享内存函数实现（复用）
int createShm(int size) {
    key_t key = ftok(PATHNAME, PROJ_ID);
    return shmget(key, size, IPC_CREAT | IPC_EXCL | 0666);
}

int getShm(int size) {
    key_t key = ftok(PATHNAME, PROJ_ID);
    return shmget(key, size, IPC_CREAT);
}

void *attachShm(int shmid) {
    return shmat(shmid, NULL, 0);
}

int detachShm(void *addr) {
    return shmdt(addr);
}

int destroyShm(int shmid) {
    return shmctl(shmid, IPC_RMID, NULL);
}

2.4.3 写进程（writer.c）

#include "sem_comm.h"
#include <unistd.h>

int main() {
    // 1. 创建信号量集（1 个信号量）并初始化（初始值 1，互斥）
    int semid = createSemSet(SEM_NUM);
    initSem(semid, 0, 1);

    // 2. 创建共享内存
    int shmid = createShm(SHM_SIZE);
    char *shmaddr = (char *)attachShm(shmid);

    // 3. 循环写入共享内存（P/V 操作保护临界区）
    for (int i = 0; i < 5; i++) {
        P(semid, 0); // 申请资源（进入临界区）
        snprintf(shmaddr, SHM_SIZE, "进程 [%d] 写入数据：%d", getpid(), i);
        printf("进程 [%d] 写入：%s\n", getpid(), shmaddr);
        V(semid, 0); // 释放资源（退出临界区）
        sleep(1);
    }

    // 4. 清理资源
    detachShm(shmaddr);
    sleep(10); // 等待读进程读取
    destroyShm(shmid);
    destroySemSet(semid);
    return 0;
}

2.4.4 读进程（reader.c）

#include "sem_comm.h"
#include <unistd.h>

int main() {
    // 1. 获取信号量集
    int semid = getSemSet(SEM_NUM);

    // 2. 获取共享内存
    int shmid = getShm(SHM_SIZE);
    char *shmaddr = (char *)attachShm(shmid);

    // 3. 循环读取共享内存（P/V 操作保护临界区）
    for (int i = 0; i < 5; i++) {
        P(semid, 0); // 申请资源（进入临界区）
        printf("进程 [%d] 读取：%s\n", getpid(), shmaddr);
        V(semid, 0); // 释放资源（退出临界区）
        sleep(1);
    }

    // 4. 清理资源
    detachShm(shmaddr);
    return 0;
}

2.4.5 运行效果

# 终端 1：启动写进程
./writer
# 输出：进程 [1234] 写入：进程 [1234] 写入数据：0
# 进程 [1234] 写入：进程 [1234] 写入数据：1

# 终端 2：启动读进程
./reader
# 输出：进程 [5678] 读取：进程 [1234] 写入数据：0
# 进程 [5678] 读取：进程 [1234] 写入数据：1

• 关键：信号量初始值为 1，确保同一时刻仅一个进程进入临界区（读写共享内存），避免数据混乱。

2.5 信号量避坑指南

• 信号量初始值设置：互斥场景初始值设为 1（二元信号量），同步场景按资源数量设置（如 2 个资源初始值设为 2）；
• P/V 操作成对出现：进入临界区前执行 P 操作，退出后执行 V 操作，避免死锁；
• 信号量集删除时机：需等待所有进程完成操作后再删除，否则正在操作的进程会报错；
• 避免信号量滥用：信号量仅用于同步互斥，不用于数据传输，不要试图通过信号量传递信息。

三、内核如何管理 System V IPC 资源

System V IPC（共享内存、消息队列、信号量）的内核管理逻辑一致，核心通过 struct ipc_ids 和 struct kern_ipc_perm 实现全局管理，这是理解 System V IPC 本质的关键。

3.1 核心管理结构

我们这里仅展示两个具有共性的管理结构体 (共享内存，消息队列，信号量都有)，剩下的具体流程看后面的部分！

3.1.1 struct ipc_ids（全局管理结构体）

内核维护三个全局 ipc_ids 结构体，分别管理共享内存、消息队列、信号量（有三个静态全局变量）：

struct ipc_ids {
    int in_use;             // 当前使用的 IPC 资源个数
    int max_id;             // 最大的 IPC 资源 ID
    unsigned short seq;     // 序列号（用于生成唯一 ID）
    unsigned short seq_max; // 序列号最大值
    struct mutex mutex;     // 保护该结构体的互斥锁
    struct ipc_id_ary nullentry; // 空条目
    struct ipc_id_ary *entries; // 指向 IPC 资源数组的指针
};

• 作用：记录系统中所有该类型 IPC 资源的元数据，实现资源的创建、查找、删除。

3.1.2 struct kern_ipc_perm（权限控制结构体）

所有 System V IPC 资源都包含 struct kern_ipc_perm 字段，用于权限控制和唯一标识：

struct kern_ipc_perm {
    spinlock_t lock;        // 自旋锁
    int deleted;            // 资源是否被标记删除
    key_t key;              // 资源的唯一 Key
    uid_t uid;              // 创建者用户 ID
    gid_t gid;              // 创建者组 ID
    uid_t cuid;             // 最后修改者用户 ID
    gid_t cgid;             // 最后修改者组 ID
    mode_t mode;            // 访问权限（如 0666）
    unsigned long seq;      // 序列号
    void *security;         // 安全相关指针
};

• 核心：key 字段是 IPC 资源的全局唯一标识，mode 字段控制进程对资源的访问权限。

3.2 内核管理流程图

3.3 重点补充：共享内存映射机制

共享内存是怎么做到把开辟出来的内存块映射到我们当前的进程地址空间的呢？这涉及到 mmap 机制。图中提到的 mmap，后续可以单独更新一篇博客来详细讲解这个知识点，这里仅仅是理解其关联关系。

3.4 关键结论和总结

• System V IPC 资源的生命周期随内核，本质是内核维护的结构体和数据结构；
• key 是资源的全局唯一标识，msgid 是进程访问资源的句柄；
• 权限控制通过 kern_ipc_perm.mode 实现，与文件权限规则一致。

本文全面覆盖了 System V 消息队列、信号量的原理、API、实战和内核管理，核心要点总结：

• 消息队列：用于结构化、异步跨进程通信，支持按类型筛选消息，适合数据传输场景；
• 信号量：用于同步与互斥，通过 P/V 操作保护临界资源，常与共享内存配合使用；
• 内核管理：System V IPC 通过 ipc_ids 和 kern_ipc_perm 实现全局管理，生命周期随内核，需手动删除；
• 选型建议：
  • 需结构化数据传输→消息队列；
  • 需同步互斥→信号量；
  • 需高效大数据传输→共享内存 + 信号量。

System V IPC 是 Linux 传统 IPC 的核心，虽然 POSIX IPC 在接口上更简洁，但 System V IPC 的设计思想（如内核管理、权限控制）仍值得深入学习。掌握这三种技术，能应对绝大多数跨进程通信场景。