在学习 Linux 进程间通信的时候，我们经常会接触到 System V IPC 这一套经典的机制，它提供了信号量、消息队列、共享内存三种方式来实现进程间的同步与通信。不过原生的系统接口往往比较繁琐，参数多、步骤杂，新手很容易踩坑。

那有没有办法把这些底层接口封装得更易用，同时还能兼顾扩展性？今天我们就结合两个实践案例，看看如何用建造者模式封装信号量，用责任链模式扩展消息队列的处理能力，把系统编程和设计模式结合起来，写出更优雅的代码。

---

一、信号量

信号量是我们实现进程间同步与互斥的基础工具，它最早由 Dijkstra 提出，核心就是通过 P、V 两个原子操作来实现对资源的控制。

1.1 核心原理

我们先回顾一下信号量的基础概念：

• 信号量本质上是一个计数器，用来表示可用资源的数量

• P 操作：申请资源，计数器减 1，如果减完后小于 0，说明资源不够，当前进程进入等待队列

• V 操作：释放资源，计数器加 1，如果加完后小于等于 0，说明有进程在等待，唤醒等待队列里的第一个进程

信号量的值的含义也很明确：

1. S > 0：表示当前有 S 个可用资源

2. S = 0：没有可用资源，也没有等待的进程

3. S < 0：绝对值表示当前等待队列里的进程数量

用伪代码来表示的话，信号量的结构体和 P、V 操作是这样的：

// 信号量结构体
struct semaphore
{
    int value;          // 计数器
    pointer_PCB queue;  // 等待进程的队列
}

// P操作：申请资源
void P(semaphore &s) {
    s.value--;
    if (s.value < 0) {
        // 资源不足，当前进程进入等待状态
        将当前进程PCB插入等待队列末尾
    }
}

// V操作：释放资源
void V(semaphore &s) {
    s.value++;
    if (s.value <= 0) {
        // 有进程在等待，唤醒一个
        从等待队列取出一个进程，放到就绪队列
    }
}

1.2 原生接口

Linux 下的 System V 信号量，提供了三个核心的系统调用，不过这些接口的参数比较多，新手很容易搞混：

semget：创建 / 获取信号量集

（1）semget：创建 / 获取信号量集

int semget(key_t key, int nsems, int semflg);

• key：信号量集的键值，用来唯一标识这个 IPC 对象，和消息队列、共享内存的 key 是同一个逻辑

• nsems：信号量集中信号量的个数，System V 信号量是支持一组信号量的

• semflg：标志位，比如IPC_CREAT表示不存在则创建，0666是权限位

（2）semctl：信号量的控制操作

这个接口是可变参数的，用来做初始化、删除等控制操作：

int semctl(int semid, int semnum, int cmd, ...);

这里有个很容易踩的坑：系统没有给我们定义 union semun 这个联合体，需要我们自己手动定义！

// 必须自己定义这个联合体
union semun {
    int val;               // 用于SETVAL，设置信号量的初始值
    struct semid_ds *buf;  // 用于IPC_STAT/IPC_SET，获取/设置属性
    unsigned short *array; // 用于GETALL/SETALL，批量获取/设置信号量值
    struct seminfo *__buf; // Linux特有的，用于获取系统信息
};

（3）semop：执行 P/V 操作

这个接口用来对信号量执行加减操作，也就是我们的 P、V 原语：

int semop(int semid, struct sembuf *sops, size_t nsops);

其中 sembuf结构体 是系统定义的，用来描述要执行的操作：

struct sembuf {
    unsigned short sem_num; // 要操作的信号量在集合中的序号
    short sem_op;           // 操作数：-1就是P操作，1就是V操作
    short sem_flg;          // 标志位，比如SEM_UNDO
};

这里的 SEM_UNDO 是个很实用的标志：当进程意外退出的时候，内核会自动撤销该进程对信号量的操作，避免进程崩溃没释放信号量导致的死锁问题，实际开发中建议加上。

1.3 封装信号量

原生的接口要创建一个信号量，需要先调用 ftok生成 key，然后semget创建信号量集，然后semctl初始化，步骤很多，而且参数分散，用起来很麻烦。

这时候建造者模式就派上用场了：建造者模式可以把复杂对象的构建和表示分离，让我们可以一步步构建对象，还能支持链式调用，让代码更清晰。

建造者模式的核心结构是这样的：

我们可以把信号量的创建步骤都封装在建造者里，对外只暴露简单的设置接口，最后调用 Build方法 就能得到一个可用的信号量对象。

（1）代码实现

我们来看看封装后的代码，首先是信号量的类，还有建造者：

#pragma once
#include <iostream>
#include <string>
#include <memory>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/sem.h>
#include <unistd.h>

const std::string pathname = "/tmp";
int proj_id = 0x77;

// 标志位：获取还是创建
#define GET_SEM IPC_CREAT
#define BUILD_SEM (IPC_CREAT | IPC_EXCL | 0666)

// 信号量类，封装P/V操作和生命周期
class Semaphore
{
public:
    Semaphore(int semid, int flag) : _semid(semid), _flag(flag) {}

    // P操作：申请锁
    void P()
    {
        struct sembuf sb;
        sb.sem_num = 0;
        sb.sem_op = -1;
        sb.sem_flg = SEM_UNDO;
        int n = ::semop(_semid, &sb, 1);
        (void)n;
    }

    // V操作：释放锁
    void V()
    {
        struct sembuf sb;
        sb.sem_num = 0;
        sb.sem_op = 1;
        sb.sem_flg = SEM_UNDO;
        int n = ::semop(_semid, &sb, 1);
        (void)n;
    }

    ~Semaphore()
    {
        if(_flag == GET_SEM) return;
        // 如果是创建者，析构的时候自动删除信号量集
        int n = ::semctl(_semid, 0, IPC_RMID);
        (void)n;
        std::cout << "sem set destroy!" << std::endl;
    }

private:
    int _semid;
    int _flag;
};

using sem_sptr = std::shared_ptr<Semaphore>;

// 建造者类，用来构建信号量对象
class SemaphoreBuilder
{
public:
    SemaphoreBuilder() : _val(-1) {}

    // 链式调用：设置信号量的初始值
    SemaphoreBuilder &SetVal(int val)
    {
        _val = val;
        return *this;
    }

    // 构建信号量对象
    sem_sptr Build(int flag)
    {
        // 合法性检查
        if (_val < 0)
        {
            std::cerr << "you must init first!" << std::endl;
            return nullptr;
        }

        // 1. 生成key
        key_t k = ::ftok(pathname.c_str(), proj_id);
        if (k < 0) exit(1);

        // 2. 创建/获取信号量集
        int semid = ::semget(k, 1, flag);
        if (semid < 0) exit(2);

        // 3. 如果是创建者，初始化信号量的初始值
        if (BUILD_SEM == flag)
        {
            union semun un;
            un.val = _val;
            int n = ::semctl(semid, 0, SETVAL, un);
            if (n < 0) exit(3);
        }

        // 4. 返回信号量对象
        return std::make_shared<Semaphore>(semid, flag);
    }

    ~SemaphoreBuilder() = default;

private:
    int _val; // 信号量的初始值
};

（2）测试

我们来测试一下这个封装好的信号量，实现父子进程交替打印，保证输出不会乱序：

#include "Sem.hpp"
#include <cstdio>
#include <time.h>
#include <unistd.h>

int main()
{
    SemaphoreBuilder sb;
    // 创建一个初始值为1的信号量，也就是二元信号量，当做互斥锁
    auto fsem = sb.SetVal(1).Build(BUILD_SEM);

    if (fork() == 0)
    {
        // 子进程：获取信号量，打印C
        auto csem = sb.Build(GET_SEM);
        int cnt = 10;
        while (cnt--)
        {
            csem->P();
            printf("C");
            fflush(stdout);
            usleep(rand() % 95270);
            printf("C ");
            usleep(rand() % 43990);
            fflush(stdout);
            csem->V();
        }
        exit(0);
    }

    // 父进程：打印F
    int cnt = 50;
    while (cnt--)
    {
        fsem->P();
        printf("F");
        fflush(stdout);
        usleep(rand() % 95270);
        printf("F ");
        usleep(rand() % 43990);
        fflush(stdout);
        fsem->V();
    }

    return 0;
}

运行之后，我们会发现输出的 CC和FF 都是成对出现的，不会出现交叉的情况，说明信号量的互斥作用生效了。

1.4 生命周期

这里要提醒大家一个很重要的点：System V 信号量的生命周期是随内核的！也就是说，就算你的进程退出了，信号量也不会自动删除，会一直留在内核里，直到你手动删除，或者系统重启。

所以我们可以用ipcs命令来查看系统里的信号量：

ipcs -s  # 查看所有信号量
ipcrm -s <semid>  # 删除指定的信号量

我们的封装里已经在析构函数里自动处理了删除的逻辑，不过如果你的进程意外崩溃，创建者没来得及析构，就需要手动用这个命令清理一下，避免资源泄露。

---

二、消息队列

说完了信号量，我们再来看看 System V 的另一个 IPC 机制：消息队列。消息队列可以让我们进程间传递有类型的数据块，比管道更灵活，支持按类型接收消息，还能实现全双工通信。

2.1 基础特性

消息队列的核心特点是：

• 每个消息都有一个类型，接收方可以根据类型选择性接收，不用按顺序读

• 自带同步机制，队列空的时候读会阻塞，队列满的时候写会阻塞

• 同样，生命周期也是随内核的

不过消息队列也有系统层面的限制，这些限制是新手很容易踩的坑，我们整理了 Linux 下的默认参数：

参数	含义	默认值
MSGMAX	单条消息的最大字节数	8192 字节
MSGMNB	单个消息队列的总字节上限	16384 字节
MSGMNI	系统中消息队列的总数上限	1730（Ubuntu 默认）

也就是说，你不能发超过 8k 的单条消息，一个队列里所有消息加起来也不能超过 16k，超过的话就会阻塞或者报错。

消息队列的通信形式很灵活，多个发送方可以给同一个队列发消息，多个接收方也可以根据自己的需求取不同类型的消息：

2.2 内核结构

在内核里，消息队列是用链表来组织的，每个队列都有一个msqid_ds的结构体来管理它的属性，里面记录了权限、时间、消息数量等信息，而消息本身则是一个链表节点，串在队列里。

内核里的结构大概是这样的：

对应的结构体定义是这样的：

struct msqid_ds {
    struct ipc_perm msg_perm;  // 权限信息
    struct msg *msg_first;     // 队列里的第一个消息
    struct msg *msg_last;      // 队列里的最后一个消息
    __kernel_time_t msg_stime; // 最后一次发送消息的时间
    __kernel_time_t msg_rtime; // 最后一次接收消息的时间
    __kernel_time_t msg_ctime; // 最后一次修改的时间
    unsigned short msg_cbytes; // 当前队列里的总字节数
    unsigned short msg_qnum;   // 当前队列里的消息数量
    unsigned short msg_qbytes; // 队列的最大字节数
    __kernel_ipc_pid_t msg_lspid; // 最后发送的进程pid
    __kernel_ipc_pid_t msg_lrpid; // 最后接收的进程pid
};

2.3 原生接口

和信号量类似，消息队列也有四个核心的接口：

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

int msgget(key_t key, int msgflg);

和semget逻辑差不多，根据 key 创建或者获取消息队列的标识符。

（2）msgctl：控制操作

int msgctl(int msqid, int cmd, struct msqid_ds *buf);

用来设置属性、删除队列等，比如IPC_RMID就是删除队列。

（4）msgsnd：发送消息

int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);

发送消息的时候，消息的结构必须以long mtype开头，也就是消息类型，后面才是数据：

struct msgbuf {
    long mtype; /* 消息类型，必须大于0 */
    char mtext[1]; /* 消息数据 */
};

（5）msgrcv：接收消息

ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp, int msgflg);

这里的msgtyp是消息队列最灵活的地方，它可以让我们选择性接收：

• msgtyp=0：接收队列里的第一条消息，不管类型

• msgtyp>0：接收队列里第一条类型等于msgtyp的消息

• msgtyp<0：接收队列里类型小于等于msgtyp绝对值的最小的消息，这就可以实现简单的优先级队列

2.4 基础的封装

首先我们先做一个基础的封装，把消息队列的收发逻辑包装起来，实现客户端和服务端的通信：

#pragma once
#include <iostream>
#include <cstring>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

#define SIZE 1024
#define PATHNAME "/tmp"
#define PROJID 0x4321

#define CREATE_NEW_MSGQUEUE (IPC_CREAT | IPC_EXCL | 0666)
#define GET_MSGQUEUE (IPC_CREAT)

// 消息结构体
typedef struct
{
    long mtype;
    char mtext[SIZE];
} msg_t;

// 消息队列基类
class MsgQueueBase
{
public:
    MsgQueueBase() = default;

    bool BuildMsgQueue(int flg)
    {
        _key = ::ftok(PATHNAME, PROJID);
        if (_key < 0) exit(1);
        _msgid = ::msgget(_key, flg);
        if (_msgid < 0) exit(2);
        return true;
    }

    // 发送消息
    bool SendMessage(const std::string &in, long type)
    {
        msg_t msg;
        msg.mtype = type;
        memset(msg.mtext, 0, sizeof(msg.mtext));
        strncpy(msg.mtext, in.c_str(), in.size());
        int n = ::msgsnd(_msgid, &msg, in.size(), 0);
        if (n < 0) return false;
        return true;
    }

    // 接收消息
    bool RecvMessage(std::string *out, long type)
    {
        msg_t msg;
        int n = ::msgrcv(_msgid, &msg, SIZE, type, 0);
        if (n < 0) return false;
        msg.mtext[n] = 0;
        *out = msg.mtext;
        return true;
    }

    // 删除消息队列
    bool DeleteMsgQueue()
    {
        int n = ::msgctl(_msgid, IPC_RMID, nullptr);
        return n == 0;
    }

    ~MsgQueueBase() = default;

protected:
    key_t _key;
    int _msgid;
};

// 客户端：获取已有的队列
class MsgQueueClient : public MsgQueueBase
{
public:
    MsgQueueClient()
    {
        bool res = MsgQueueBase::BuildMsgQueue(GET_MSGQUEUE);
        (void)res;
    }
};

// 服务端：创建新的队列
class MsgQueueServer : public MsgQueueBase
{
public:
    MsgQueueServer()
    {
        bool res = MsgQueueBase::BuildMsgQueue(CREATE_NEW_MSGQUEUE);
        (void)res;
    }

    ~MsgQueueServer()
    {
        bool res = MsgQueueBase::DeleteMsgQueue();
        (void)res;
    }
};

#define SERVER 1
#define CLIENT 2

然后客户端和服务端的代码就很简单了：

// 客户端
#include "MsgQueue.hpp"
int main()
{
    std::string msg = "hello bite";
    MsgQueueClient mq;
    mq.SendMessage(msg, CLIENT);
    return 0;
}

// 服务端
#include "MsgQueue.hpp"
int main()
{
    std::string msg;
    MsgQueueServer mq;
    mq.RecvMessage(&msg, CLIENT);
    std::cout << "get message: " << msg << std::endl;
    return 0;
}

这个基础版本可以实现简单的通信，但是如果我们有新的需求呢？

2.5 扩展消息处理

比如我们现在有了新的需求：

1. 收到的消息，要加上时间戳和进程 pid，方便排查问题

2. 处理后的消息要持久化保存到日志文件里

3. 如果日志文件太大了，要自动备份打包，避免文件过大

如果我们把这些逻辑都硬编码到接收消息的函数里，那代码会越来越臃肿，而且如果我有时候不需要备份，有时候不需要加时间戳，改起来也很麻烦，耦合度太高了。

这时候责任链模式就派上用场了！责任链模式是一种行为设计模式，它可以让我们把请求的处理者连成一条链，每个处理者都可以选择自己处理这个请求，或者把它传给下一个处理者。这样就把发送者和接收者解耦了，每个处理者都是独立的，可以随时开关，也可以随时新增。

责任链模式的结构是这样的：

每个处理者都有一个指向下一个处理者的指针，请求过来的时候，先自己处理，然后如果有下一个，就传给下一个，这样就把每个处理步骤都解耦开了。

（1）代码实现

我们来实现这个责任链，首先是处理者的基类：

#pragma once
#include <iostream>
#include <filesystem>
#include <memory>
#include <unistd.h>
#include <sstream>
#include <fstream>
#include <ctime>
#include <sys/types.h>
#include <sys/wait.h>

// 处理者基类
class HandlerText
{
public:
    HandlerText() : _enable(true) {}
    virtual ~HandlerText() = default;

    // 设置下一个处理者
    void SetNextHandler(std::shared_ptr<HandlerText> handler)
    {
        _next_handler = handler;
    }

    // 开关：是否启用这个处理者
    void Enable() { _enable = true; }
    void DisEnable() { _enable = false; }
    bool IsEnable() { return _enable; }

    // 执行处理，子类实现
    virtual void Execute(std::string &info) = 0;

protected:
    std::shared_ptr<HandlerText> _next_handler;
    bool _enable;
};

然后我们实现三个具体的处理者，分别对应我们的三个需求：

第一个：格式化处理，给消息加上时间和 pid：

// 格式化处理：加时间戳和pid
class HandlerTextFormat : public HandlerText
{
public:
    HandlerTextFormat() = default;

    void Execute(std::string &info) override
    {
        if (HandlerText::IsEnable())
        {
            std::cout << "Format ..." << std::endl;
            // 拼接时间和pid
            std::stringstream ss;
            ss << time(nullptr) << " - " << getpid() << " - " << info << "\n";
            info = ss.str();
            sleep(1);
        }

        // 传给下一个处理者
        if (_next_handler)
            _next_handler->Execute(info);
        else
            std::cout << "责任链节点结束,处理完成" << std::endl;
    }
};

第二个：保存到文件：

std::string defaultpath = "./tmp/";
std::string defaultfilename = "test.log";

// 保存到文件
class HandlerTextSaveFile : public HandlerText
{
public:
    HandlerTextSaveFile() : _filepath(defaultpath), _filename(defaultfilename)
    {
        // 自动创建目录
        if (std::filesystem::exists(_filepath))
            return;
        try
        {
            std::filesystem::create_directories(_filepath);
        }
        catch (std::filesystem::filesystem_error &e)
        {
            std::cerr << e.what() << std::endl;
        }
    }
    ~HandlerTextSaveFile() = default;

    void Execute(std::string &info) override
    {
        if (HandlerText::IsEnable())
        {
            std::cout << "Save ..." << std::endl;
            sleep(1);
            // 追加写入文件
            const std::string file = _filepath + _filename;
            std::ofstream out(file, std::ios::app);
            if (!out.is_open())
                return;
            out << info;
            out.close();
        }

        if (_next_handler)
            _next_handler->Execute(info);
        else
            std::cout << "责任链节点结束,处理完成" << std::endl;
    }

private:
    std::string _filepath;
    std::string _filename;
};

第三个：备份文件，当文件行数超过阈值的时候，自动打包备份：

const int maxline = 5; // 测试用的阈值，实际可以调大

// 备份处理：文件过大自动打包
class HandlerTextBackupFile : public HandlerText
{
public:
    HandlerTextBackupFile() : _max_line_number(maxline), 
                                _filepath(defaultpath), _filename(defaultfilename) {}
    ~HandlerTextBackupFile() = default;

    void Execute(std::string &info) override
    {
        if (HandlerText::IsEnable())
        {
            std::cout << "Backup ..." << std::endl;
            sleep(1);
            const std::string filename = _filepath + _filename;
            
            // 统计当前文件的行数
            std::ifstream in(filename);
            if (!in.is_open()) return;
            int currentlines = 0;
            std::string line;
            while (std::getline(in, line))
            {
                currentlines++;
            }
            in.close();

            // 超过阈值就备份
            if (currentlines > _max_line_number)
            {
                std::cout << "消息行数超过" << _max_line_number << ", 触发日志备份" << std::endl;
                Backup();
            }
        }

        if (_next_handler)
            _next_handler->Execute(info);
        else
            std::cout << "责任链节点结束,处理完成" << std::endl;
    }

private:
    // 备份逻辑：重命名然后打包
    void Backup()
    {
        std::string newname = _filename + "." + std::to_string(time(nullptr));
        pid_t id = fork();
        if (id == 0)
        {
            chdir(_filepath.c_str());
            // 重命名原文件
            std::filesystem::rename(_filename, newname);
            std::string tarname = newname + ".tgz";
            // 子进程执行tar打包
            execlp("tar", "tar", "czf", tarname.c_str(), newname.c_str(), nullptr);
            exit(1);
        }
        waitpid(id, nullptr, 0);
        // 删除原文件，只保留压缩包
        std::string tempfile = _filepath + newname;
        std::filesystem::remove(tempfile);
    }

    int _max_line_number;
    std::string _filepath;
    std::string _filename;
};

最后，我们做一个入口类，用来构建责任链，对外提供统一的调用接口：

class HandlerEntry
{
public:
    HandlerEntry()
    {
        // 创建所有处理者
        _format = std::make_shared<HandlerTextFormat>();
        _save = std::make_shared<HandlerTextSaveFile>();
        _backup = std::make_shared<HandlerTextBackupFile>();

        // 把处理者连成链
        _format->SetNextHandler(_save);
        _save->SetNextHandler(_backup);
    }

    // 控制每个处理者的开关
    void EnableHandler(bool isformat, bool issave, bool isbackup)
    {
        isformat ? _format->Enable() : _format->DisEnable();
        issave ? _save->Enable() : _save->DisEnable();
        isbackup ? _backup->Enable() : _backup->DisEnable();
    }

    // 执行处理
    void Run(std::string &info)
    {
        _format->Execute(info);
    }

private:
    std::shared_ptr<HandlerText> _format;
    std::shared_ptr<HandlerText> _save;
    std::shared_ptr<HandlerText> _backup;
};

（2）优势

这样封装之后，我们的代码就变得非常灵活了：

• 每个处理步骤都是独立的，改格式化的逻辑不会影响保存，改备份的逻辑也不会影响其他的

• 可以随时开关某个步骤，比如我今天不需要备份，只需要调用EnableHandler(true, true, false)就可以了，不用改任何代码

• 新增处理步骤也很简单，比如我要加一个过滤敏感词的处理，只需要新增一个HandlerTextFilter的类，然后加到链里就可以了，完全符合开闭原则

这就是设计模式的魅力，把复杂的逻辑拆分成独立的模块，让代码更容易维护和扩展。

2.6 生命周期

和信号量一样，消息队列的生命周期也是随内核的，所以我们也可以用ipcs命令来管理：

ipcs -q  # 查看所有消息队列
ipcrm -q <msgid>  # 删除指定的消息队列

同样，我们的封装里也自动处理了删除的逻辑，不过如果进程崩溃了，还是需要手动清理一下。

---

三、总结

今天我们结合两个实践案例，把 System V IPC 的底层接口和设计模式结合了起来：

• 用建造者模式封装了信号量，把复杂的创建步骤隐藏起来，对外提供了简洁的链式调用接口，让信号量的使用变得简单

• 用责任链模式扩展了消息队列的处理能力，把消息的多个处理步骤解耦，让代码的扩展性大大提升

其实设计模式并不是只有在业务开发里才有用，在系统编程里，同样可以用设计模式来封装底层的复杂接口，让我们的代码更优雅、更容易维护。对于我们计算机专业的同学来说，把系统编程的知识和面向对象的设计思想结合起来，才能写出真正高质量的代码。

如果你在实践中遇到了 IPC 相关的问题，或者对设计模式的应用有其他的想法，欢迎在评论区交流～