消息队列原理与实战：Linux IPC 接口及责任链模式设计

基础封装类

#include <sys/msg.h>
#include <sys/ipc.h>
#include <cstring>
#include <iostream>
#include <string>
#include <unistd.h>
#include <cstdlib>

using namespace std;

// 定义消息结构体
struct MsgBuf {
    long mtype;
    char mtext[1024];
};

class MsgQueue {
protected:
    int key_;
    int msgid_;

public:
    MsgQueue(const string &pathname, int proj_id, int mode) {
        key_ = ftok(pathname.c_str(), proj_id);
        if (key_ < 0) {
            cerr << "ftok error" << endl;
            exit(EXIT_FAILURE);
        }
        msgid_ = msgget(key_, mode);
        if (msgid_ < 0) {
            cerr << "msgget error" << endl;
            exit(EXIT_FAILURE);
        }
    }

    ~MsgQueue() {
        // 析构时不自动删除，需显式调用 cleanup
    }

    bool Send(const string &message, long mtype) {
        MsgBuf buffer;
        memset(buffer.mtext, 0, sizeof(buffer.mtext));
        buffer.mtype = mtype;
        memcpy(buffer.mtext, message.c_str(), message.size());
        
        if (msgsnd(msgid_, &buffer, sizeof(buffer.mtext), 0) < 0) {
            cerr << "msgsnd error" << endl;
            return false;
        }
        return true;
    }

    string Recv(long mtype) {
        MsgBuf buffer;
        memset(buffer.mtext, 0, sizeof(buffer.mtext));
        ssize_t n = msgrcv(msgid_, &buffer, sizeof(buffer.mtext), mtype, 0);
        if (n < 0) {
            cerr << "msgrcv error" << endl;
            return "";
        }
        return string(buffer.mtext);
    }
};

服务端与客户端

服务端负责创建队列，客户端仅连接。注意 mode 参数的区别：服务端使用 IPC_CREAT | IPC_EXCL，客户端使用 IPC_CREAT。

const int CREATE_MODE = IPC_CREAT | IPC_EXCL | 0666;
const int JOIN_MODE   = IPC_CREAT;

class Server : public MsgQueue {
public:
    Server(const string &path = "/tmp", int id = 123) 
        : MsgQueue(path, id, CREATE_MODE) {}
    
    ~Server() {
        msgctl(msgid_, IPC_RMID, nullptr);
    }
};

class Client : public MsgQueue {
public:
    Client(const string &path = "/tmp", int id = 123) 
        : MsgQueue(path, id, JOIN_MODE) {}
};

主程序中利用 fork() 模拟父子进程通信，一方发送，另一方接收，通过特定字符串（如 "quit"）结束循环。

进阶：责任链模式处理消息

当消息处理逻辑变得复杂时（例如：校验 -> 日志 -> 存储 -> 归档），硬编码在同一个函数里会导致代码臃肿且难以扩展。此时，责任链模式（Chain of Responsibility）是绝佳选择。

设计思路

我们将处理步骤拆分为独立的处理器，每个处理器持有对下一个处理器的引用。请求沿着链条传递，直到被处理完毕或被中断。

基类定义

#include <memory>
#include <functional>

class Handler {
public:
    virtual void Execute(const string &message) = 0;
    
    void SetNext(std::shared_ptr<Handler> handler) {
        next_handler_ = handler;
    }

protected:
    std::shared_ptr<Handler> next_handler_;
    bool is_open_ = true; // 控制是否启用此节点
};

具体处理器

1. 时间戳封装：添加时间和 PID 信息。
2. 文件保存：写入磁盘，自动创建目录。
3. 压缩归档：当文件行数超过阈值时，触发 tar 打包。

以压缩处理器为例，它需要判断文件大小，若超标则调用系统命令进行压缩，并删除源文件。

class CompressHandler : public Handler {
private:
    string filepath_;
    string filename_;
    int max_line_;

public:
    CompressHandler(const string &fp, const string &fn, int max) 
        : filepath_(fp), filename_(fn), max_line_(max) {}

    void Execute(const string &message) override {
        if (!is_open_) {
            if (next_handler_) next_handler_->Execute(message);
            return;
        }

        // 检查是否需要压缩
        string file = filepath_ + "/" + filename_;
        // ... 省略具体的行计数逻辑 ...
        
        // 假设需要压缩，执行 tar 命令
        pid_t pid = fork();
        if (pid == 0) {
            // 子进程执行压缩
            execlp("tar", "tar", "-zcf", "backup.tar.gz", filename_.c_str(), NULL);
            exit(1);
        } else {
            // 父进程等待并清理
            wait(nullptr);
            remove(file.c_str());
        }
        
        if (next_handler_) next_handler_->Execute(message);
    }
};

串联运行

最后，在主流程中将各个处理器串联起来：

auto h1 = make_shared<TimeStampHandler>();
auto h2 = make_shared<FileSaveHandler>("/data", "log.txt");
auto h3 = make_shared<CompressHandler>("/data", "log.txt", 1000);

h1->SetNext(h2);
h2->SetNext(h3);

h1->Execute("Hello World");

这种设计使得新增处理步骤（如加密、过滤）无需修改现有代码，只需插入新节点即可，极大地提升了系统的灵活性和可维护性。