尧图网页设计 
目录一、消息队列1、概述2、通信形式3、IPC数据结构4、消息队列数据结构5、内核表示6、接口1msgget2msgctl3msgsnd4msgrcv7、通信实现二、责任链模式1、概述2、实现三、结语一、消息队列1、概述消息队列提供了一个从一个进程向另外一个进程发送有类型块数据的方法。每个数据块都被认为是有一个类型接收者进程接收的数据块可以有不同的类型值。消息队列也有管道一样的不足就是每个消息的最大长度是有上限的。每个消息队列的总的字节数也是有上限的系统上消息队列的总数也有上限。2、通信形式上图展示的就是Linux内核中消息队列的通信机制。在用户态有两个进程进程1和进程2分别关注不同类型的消息内核态维护着一个消息队列进程发送的消息按类型和内容依次排入队列。进程1只从队列中取走消息类型1的消息进程2只取走消息类型2的消息箭头代表的就是这种按类型筛选接收的过程。整个机制体现了消息队列的异步通信和类型过滤能力发送方与接收方无需直接交互消息由内核中转并按类型分发。3、IPC数据结构在/usr/include/linux/ipc.h中内核为每个IPC对象维护一个数据结构如下所示struct ipc_perm{ key_t __key; /* Key supplied to xxxget(2) */ uid_t uid; /* Effective UID of owner */ gid_t gid; /* Effective GID of owner */ uid_t cuid; /* Effective UID of creator */ gid_t cgid; /* Effective GID of creator */ unsigned short mode; /* Permissions */ unsigned short __seq; /* Sequence number */ };以上代码是Linux内核中用于描述IPC对象权限的结构体ipc_perm常见于消息队列、共享内存、信号量等System V IPC机制中。它定义了某个IPC资源的所有者、创建者以及访问权限等元信息。其中_key是该IPC对象对外使用的键值用于不同进程找到同一个资源uid和gid分别表示当前所有者的有效用户ID和组IDcuid和cgid表示创建者的用户ID和组IDmode是权限位决定哪些用户或组可以读、写或执行相关操作_seq是序列号内核用它来复用IPC标识符时避免冲突。4、消息队列数据结构在/usr/include/linux/msg.h中描述消息队列信息的结构体如下所示struct msqid_ds { struct ipc_perm msg_perm; struct msg msg_first; / first message on queue,unused */ struct msg msg_last; / last message in queue,unused */ __kernel_time_t msg_stime; /* last msgsnd time */ __kernel_time_t msg_rtime; /* last msgrcv time */ __kernel_time_t msg_ctime; /* last change time */ unsigned long msg_lcbytes; /* Reuse junk fields for 32 bit */ unsigned long msg_lqbytes; /* ditto */ unsigned short msg_cbytes; /* current number of bytes on queue */ unsigned short msg_qnum; /* number of messages in queue */ unsigned short msg_qbytes; /* max number of bytes on queue */ __kernel_ipc_pid_t msg_lspid; /* pid of last msgsnd */ __kernel_ipc_pid_t msg_lrpid; /* last receive pid */ };以上代码是Linux内核中描述消息队列实例的结构体msqid_ds它对应一个具体的消息队列对象记录了队列的权限、状态、统计信息和时间戳。msg_perm负责权限和所有者信息msg_first和msg_last是队列头尾指针msg_stime、msg_rtime、msg_ctime分别记录最后一次发送、接收和修改的时间msg_cbytes表示当前队列中已占用的字节数msg_qnum是当前消息条数msg_qbytes是队列允许的最大字节数msg_lspid和msg_lrpid记录最后一次发送和接收消息的进程ID。5、内核表示下图展示了Linux内核中System V消息队列的完整数据结构关系右侧msg_queue是队列的管理结构包含权限、时间戳、当前字节数、消息条数、最大字节数以及等待发送和接收的进程。中间msg_msg是消息节点通过双向链表挂在q_message上每个节点记录消息类型和大小并指向真正的消息正文。下方的msg_send和msg_rcv分别代表发送和接收进程通过指针与消息正文关联。6、接口1msggetmsgget用于System V消息队列的创建或获取系统调用。msgget的功能是根据一个键值key在内核中查找已有的消息队列如果找不到就创建一个新的并返回该队列的标识符msqid。参数key某个消息队列的名字msgflg由九个权限标志构成它们的用法和创建文件时使用的mode模式标志是一样的返回值成功返回一个非负整数即该消息队列的标识码失败返回-1。2msgctlmsgctl是System V消息队列的控制系统调用。它通过msqid定位到具体的消息队列然后根据cmd参数执行不同的管理操作。参数msgid由msgget函数返回的消息队列标识码。cmd将要采取的动作有三个可取值如下图所示buf属性缓冲区。返回值成功返回0失败返回-1。3msgsndmsgsnd是System V消息队列的发送消息系统调用。它把一条消息写入指定的消息队列如果队列已满或权限不足则可能阻塞或立即返回错误。参数msgid由msgget函数返回的消息队列标识码msgp是一个指针指针指向准备发送的消息msgsz是msgp指向的消息长度这个长度不含保存消息类型的那个long int长整型msgflg控制着当前消息队列满或到达系统上限时将要发生的事情。返回值成功返回0失败返回-1。消息主体struct msgbuf { long mtype; /* message type, must be 0 */ char mtext[1]; /* message data */ };消息主体以一个long int长整型开始接受者函数将利用这个确定消息的类型。4msgrcvmsgrcv是System V消息队列的接收消息系统调用它从指定队列中取走一条消息并可以按消息类型进行筛选。参数msgid由msgget函数返回的消息队列标识码msgp是一个指针指针指向准备接收的消息msgsz是msgp指向的消息长度这个长度不含保存消息类型的那个long int长整型msgtyp它可以实现接收消息的类型也可以模拟优先级的简单形式进行接收msgtyp为0时返回队列中第一条消息大于0时返回第一条类型等于msgtyp的消息小于0时返回第一条类型小于等于msgtyp绝对值、且满足条件中类型最小的那条消息。msgflg控制着队列中没有相应类型的消息可供接收时将要发生的事。IPC_NOWAIT表示队列无可读消息时不等待直接返回ENOMSG错误MSG_NOERROR表示消息正文超过msgsz时自动截断而不报错。此外当msgtyp大于0且msgflg为MSG_EXCEPT时接收的是类型不等于msgtyp的第一条消息相当于对指定类型做排除。返回值成功返回实际放到接收缓冲区里的字符个数失败返回-1。7、通信实现msgqueue.hpp#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() { } 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() { } 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以上代码封装了一个基于System V消息队列的C简易通信类。其中MsgQueueBase负责公共逻辑ftok用于生成键值、msgget用于创建或获取队列、msgsnd发送、msgrcv接收、msgctl删除。MsgQueueClient和MsgQueueServer分别继承它客户端用IPC_CREAT获取已有队列服务端用IPC_CREAT | IPC_EXCL | 0666创建新队列服务端析构时删除队列。server.cc#include msgqueue.hpp int main() { std::string msg; MsgQueueServer mq; mq.RecvMessage(msg, CLIENT); std::cout get message: msg std::endl; return 0; }client.cc#include msgqueue.hpp int main() { std::string msg hello msgqueue; MsgQueueClient mq; mq.SendMessage(msg, CLIENT); return 0; }Makefile.PHONY:all all:client server client:client.cc msgqueue.hpp g -o $ client.cc -stdc11 server:server.cc msgqueue.hpp g -o $ server.cc -stdc11 .PHONY:clean clean: rm -f client server执行make编译生成client、server可执行文件./server服务端先启动进行阻塞等待在另一个终端启动客户端./client客户端发送消息后服务端将打印结果如下图所示如上图所示服务端成功打印消息get messagehello msgqueue。至此客户端和服务端二者通过消息队列成功实现通信。需要注意的是消息队列的生命周期是随内核的消息队列支持全双工通信。二、责任链模式1、概述新需求client发送给server的输入内容拼接上时间进程pid信息server收到的内容持久化保存到文件中文件的内容如果过大要进行切片保存并在指定的目录下打包保存命令自定义要解决这个需求就得用到责任链模式责任链模式是一种行为设计模式它允许请求沿着处理者链进行传递。每个处理者都对请求进行检查以决定是否处理它。如果处理者能够处理该请求就处理它否则它将请求传递给链中的下一个处理者。这个模式使得多个对象都有机会处理请求从而避免了请求的发送者和接收者之间的紧耦合后续新增或调整处理环节只需改动链的组装不用动客户端和已有处理器。2、实现#pragma once #includeiostream #includefilesystem #includememory #includeunistd.h #includesstream #includefstream #includectime #includesys/types.h #includesys/wait.h class HandlerText { public: HandlerText() : _enable(true) { } virtual ~HandlerText() default; void SetNextHandler(std::shared_ptrHandlerText handler) { _next_handler handler; } void Enable() { _enable true; } void DisEnable() { _enable false; } bool IsEnable() { return _enable; } virtual void Execute(std::string info) 0; protected: // 这⾥要protected⽅便继承 std::shared_ptrHandlerText _next_handler; bool _enable; }; class HandlerTextFormat : public HandlerText { public: ~HandlerTextFormat() { } void Execute(std::string info) override { if (HandlerText::IsEnable()) { // 开始处理添加简单的补充信息 std::cout Format ... std::endl; // 简单处理 std::stringstream ss; ss time(nullptr) - getpid() - info \n; info ss.str(); sleep(1); } if (_next_handler) // 如果_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() { } 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被设置就交给下⼀个继续加⼯处理 _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() { } void Execute(std::string info) override { if (HandlerText::IsEnable()) { // 开始处理对⽂件进⾏增量备份 std::cout Backup ... std::endl; sleep(1); const std::string filename _filepath _filename; // 1. 打开⽂件 std::ifstream in(filename); if (!in.is_open()) return; std::string line; int currentlines 0; while (std::getline(in, line)) { currentlines; } // 关闭⽂件流 in.close(); // 2. 备份 if (currentlines _max_line_number) { std::cout 消息⾏数超过 _max_line_number , 触发⽇志备 份 std::endl; // ⼤于才做备份否则什么⾛不做 Backup(); } } if (_next_handler) // 如果_next_handler被设置就交给下⼀个继续加⼯处理 _next_handler-Execute(info); else std::cout 责任链节点结束,处理完成 std::endl; } void Backup() { std::string newname _filename . std::to_string(time(nullptr)); pid_t id fork(); if (id 0) { chdir(_filepath.c_str()); // 更改进程路径进⼊./tmp/路径下 std::filesystem::rename(_filename, newname); // rename⽐较快也不影 响未来其他继续写⼊的操作因为会重新形成⽂件 std::string tarname newname .tgz; // ⼦进程打包备份 std::cout 打包 : newname 成为: tarname 开始 std::endl; execlp(tar, tar, czf, tarname.c_str(), newname.c_str(), nullptr); // 注意这⾥要以nullptr结尾注意这⾥的坑 std::cout 打包 : newname 成为: tarname 失败 std::endl; exit(1); } waitpid(id, nullptr, 0); std::string tempfile _filepath newname; std::filesystem::remove(tempfile); // 删除⽂件原件只要tar包 } private: int _max_line_number; std::string _filepath; std::string _filename; }; class HandlerEntry { public: HandlerEntry() { // 构建责任链节点对象 _format std::make_sharedHandlerTextFormat(); _save std::make_sharedHandlerTextSaveFile(); _backup std::make_sharedHandlerTextBackupFile(); // 链接责任链 _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_ptrHandlerText _format; std::shared_ptrHandlerText _save; std::shared_ptrHandlerText _backup; };以上代码采用责任链模式实现了一个文本处理流水线。HandlerText是抽象基类持有后继节点和启用开关三个具体处理器分别负责格式化文本、追加写入日志文件、按行数阈值触发备份。HandlerTextFormat给文本加上时间戳和进程号HandlerTextSaveFile把结果写到./tmp/test.logHandlerTextBackupFile检查行数超限后fork子进程用tar打包备份并删除原文件。HandlerEntry负责把三个节点按顺序组装成链并提供统一开关和启动入口。整体思路是请求从链头进入依次经过各节点处理每个节点只关心自己的逻辑处理完交给下一个实现了解耦和可插拔。三、结语本文从System V消息队列的底层机制出发从核心接口和内核数据结构切入最后通过一个责任链模式的文本处理流水线作为实践。消息队列解决的是进程间怎么传的问题责任链解决的是消费端怎么处理的问题。前者让消息异步流动后者让处理逻辑可插拔、可编排。把两者结合消费端就不再是一个巨大的if-else方法而是一条清晰的处理链消息从队列取出后依次经过格式化、落盘、备份等节点每个节点只关心自己的逻辑新增或调整环节只需改动链的组装。用消息队列解决进程间通信的解耦用责任链解决消费端处理逻辑的解耦两者结合就能实现让异步任务从能收发走向可编排。 基于责任链模式的消息队列—异步处理流水线的最佳实践
目录一、消息队列1、概述2、通信形式3、IPC数据结构4、消息队列数据结构5、内核表示6、接口1msgget2msgctl3msgsnd4msgrcv7、通信实现二、责任链模式1、概述2、实现三、结语一、消息队列1、概述消息队列提供了一个从一个进程向另外一个进程发送有类型块数据的方法。每个数据块都被认为是有一个类型接收者进程接收的数据块可以有不同的类型值。消息队列也有管道一样的不足就是每个消息的最大长度是有上限的。每个消息队列的总的字节数也是有上限的系统上消息队列的总数也有上限。2、通信形式上图展示的就是Linux内核中消息队列的通信机制。在用户态有两个进程进程1和进程2分别关注不同类型的消息内核态维护着一个消息队列进程发送的消息按类型和内容依次排入队列。进程1只从队列中取走消息类型1的消息进程2只取走消息类型2的消息箭头代表的就是这种按类型筛选接收的过程。整个机制体现了消息队列的异步通信和类型过滤能力发送方与接收方无需直接交互消息由内核中转并按类型分发。3、IPC数据结构在/usr/include/linux/ipc.h中内核为每个IPC对象维护一个数据结构如下所示struct ipc_perm{ key_t __key; /* Key supplied to xxxget(2) */ uid_t uid; /* Effective UID of owner */ gid_t gid; /* Effective GID of owner */ uid_t cuid; /* Effective UID of creator */ gid_t cgid; /* Effective GID of creator */ unsigned short mode; /* Permissions */ unsigned short __seq; /* Sequence number */ };以上代码是Linux内核中用于描述IPC对象权限的结构体ipc_perm常见于消息队列、共享内存、信号量等System V IPC机制中。它定义了某个IPC资源的所有者、创建者以及访问权限等元信息。其中_key是该IPC对象对外使用的键值用于不同进程找到同一个资源uid和gid分别表示当前所有者的有效用户ID和组IDcuid和cgid表示创建者的用户ID和组IDmode是权限位决定哪些用户或组可以读、写或执行相关操作_seq是序列号内核用它来复用IPC标识符时避免冲突。4、消息队列数据结构在/usr/include/linux/msg.h中描述消息队列信息的结构体如下所示struct msqid_ds { struct ipc_perm msg_perm; struct msg msg_first; / first message on queue,unused */ struct msg msg_last; / last message in queue,unused */ __kernel_time_t msg_stime; /* last msgsnd time */ __kernel_time_t msg_rtime; /* last msgrcv time */ __kernel_time_t msg_ctime; /* last change time */ unsigned long msg_lcbytes; /* Reuse junk fields for 32 bit */ unsigned long msg_lqbytes; /* ditto */ unsigned short msg_cbytes; /* current number of bytes on queue */ unsigned short msg_qnum; /* number of messages in queue */ unsigned short msg_qbytes; /* max number of bytes on queue */ __kernel_ipc_pid_t msg_lspid; /* pid of last msgsnd */ __kernel_ipc_pid_t msg_lrpid; /* last receive pid */ };以上代码是Linux内核中描述消息队列实例的结构体msqid_ds它对应一个具体的消息队列对象记录了队列的权限、状态、统计信息和时间戳。msg_perm负责权限和所有者信息msg_first和msg_last是队列头尾指针msg_stime、msg_rtime、msg_ctime分别记录最后一次发送、接收和修改的时间msg_cbytes表示当前队列中已占用的字节数msg_qnum是当前消息条数msg_qbytes是队列允许的最大字节数msg_lspid和msg_lrpid记录最后一次发送和接收消息的进程ID。5、内核表示下图展示了Linux内核中System V消息队列的完整数据结构关系右侧msg_queue是队列的管理结构包含权限、时间戳、当前字节数、消息条数、最大字节数以及等待发送和接收的进程。中间msg_msg是消息节点通过双向链表挂在q_message上每个节点记录消息类型和大小并指向真正的消息正文。下方的msg_send和msg_rcv分别代表发送和接收进程通过指针与消息正文关联。6、接口1msggetmsgget用于System V消息队列的创建或获取系统调用。msgget的功能是根据一个键值key在内核中查找已有的消息队列如果找不到就创建一个新的并返回该队列的标识符msqid。参数key某个消息队列的名字msgflg由九个权限标志构成它们的用法和创建文件时使用的mode模式标志是一样的返回值成功返回一个非负整数即该消息队列的标识码失败返回-1。2msgctlmsgctl是System V消息队列的控制系统调用。它通过msqid定位到具体的消息队列然后根据cmd参数执行不同的管理操作。参数msgid由msgget函数返回的消息队列标识码。cmd将要采取的动作有三个可取值如下图所示buf属性缓冲区。返回值成功返回0失败返回-1。3msgsndmsgsnd是System V消息队列的发送消息系统调用。它把一条消息写入指定的消息队列如果队列已满或权限不足则可能阻塞或立即返回错误。参数msgid由msgget函数返回的消息队列标识码msgp是一个指针指针指向准备发送的消息msgsz是msgp指向的消息长度这个长度不含保存消息类型的那个long int长整型msgflg控制着当前消息队列满或到达系统上限时将要发生的事情。返回值成功返回0失败返回-1。消息主体struct msgbuf { long mtype; /* message type, must be 0 */ char mtext[1]; /* message data */ };消息主体以一个long int长整型开始接受者函数将利用这个确定消息的类型。4msgrcvmsgrcv是System V消息队列的接收消息系统调用它从指定队列中取走一条消息并可以按消息类型进行筛选。参数msgid由msgget函数返回的消息队列标识码msgp是一个指针指针指向准备接收的消息msgsz是msgp指向的消息长度这个长度不含保存消息类型的那个long int长整型msgtyp它可以实现接收消息的类型也可以模拟优先级的简单形式进行接收msgtyp为0时返回队列中第一条消息大于0时返回第一条类型等于msgtyp的消息小于0时返回第一条类型小于等于msgtyp绝对值、且满足条件中类型最小的那条消息。msgflg控制着队列中没有相应类型的消息可供接收时将要发生的事。IPC_NOWAIT表示队列无可读消息时不等待直接返回ENOMSG错误MSG_NOERROR表示消息正文超过msgsz时自动截断而不报错。此外当msgtyp大于0且msgflg为MSG_EXCEPT时接收的是类型不等于msgtyp的第一条消息相当于对指定类型做排除。返回值成功返回实际放到接收缓冲区里的字符个数失败返回-1。7、通信实现msgqueue.hpp#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() { } 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() { } 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以上代码封装了一个基于System V消息队列的C简易通信类。其中MsgQueueBase负责公共逻辑ftok用于生成键值、msgget用于创建或获取队列、msgsnd发送、msgrcv接收、msgctl删除。MsgQueueClient和MsgQueueServer分别继承它客户端用IPC_CREAT获取已有队列服务端用IPC_CREAT | IPC_EXCL | 0666创建新队列服务端析构时删除队列。server.cc#include msgqueue.hpp int main() { std::string msg; MsgQueueServer mq; mq.RecvMessage(msg, CLIENT); std::cout get message: msg std::endl; return 0; }client.cc#include msgqueue.hpp int main() { std::string msg hello msgqueue; MsgQueueClient mq; mq.SendMessage(msg, CLIENT); return 0; }Makefile.PHONY:all all:client server client:client.cc msgqueue.hpp g -o $ client.cc -stdc11 server:server.cc msgqueue.hpp g -o $ server.cc -stdc11 .PHONY:clean clean: rm -f client server执行make编译生成client、server可执行文件./server服务端先启动进行阻塞等待在另一个终端启动客户端./client客户端发送消息后服务端将打印结果如下图所示如上图所示服务端成功打印消息get messagehello msgqueue。至此客户端和服务端二者通过消息队列成功实现通信。需要注意的是消息队列的生命周期是随内核的消息队列支持全双工通信。二、责任链模式1、概述新需求client发送给server的输入内容拼接上时间进程pid信息server收到的内容持久化保存到文件中文件的内容如果过大要进行切片保存并在指定的目录下打包保存命令自定义要解决这个需求就得用到责任链模式责任链模式是一种行为设计模式它允许请求沿着处理者链进行传递。每个处理者都对请求进行检查以决定是否处理它。如果处理者能够处理该请求就处理它否则它将请求传递给链中的下一个处理者。这个模式使得多个对象都有机会处理请求从而避免了请求的发送者和接收者之间的紧耦合后续新增或调整处理环节只需改动链的组装不用动客户端和已有处理器。2、实现#pragma once #includeiostream #includefilesystem #includememory #includeunistd.h #includesstream #includefstream #includectime #includesys/types.h #includesys/wait.h class HandlerText { public: HandlerText() : _enable(true) { } virtual ~HandlerText() default; void SetNextHandler(std::shared_ptrHandlerText handler) { _next_handler handler; } void Enable() { _enable true; } void DisEnable() { _enable false; } bool IsEnable() { return _enable; } virtual void Execute(std::string info) 0; protected: // 这⾥要protected⽅便继承 std::shared_ptrHandlerText _next_handler; bool _enable; }; class HandlerTextFormat : public HandlerText { public: ~HandlerTextFormat() { } void Execute(std::string info) override { if (HandlerText::IsEnable()) { // 开始处理添加简单的补充信息 std::cout Format ... std::endl; // 简单处理 std::stringstream ss; ss time(nullptr) - getpid() - info \n; info ss.str(); sleep(1); } if (_next_handler) // 如果_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() { } 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被设置就交给下⼀个继续加⼯处理 _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() { } void Execute(std::string info) override { if (HandlerText::IsEnable()) { // 开始处理对⽂件进⾏增量备份 std::cout Backup ... std::endl; sleep(1); const std::string filename _filepath _filename; // 1. 打开⽂件 std::ifstream in(filename); if (!in.is_open()) return; std::string line; int currentlines 0; while (std::getline(in, line)) { currentlines; } // 关闭⽂件流 in.close(); // 2. 备份 if (currentlines _max_line_number) { std::cout 消息⾏数超过 _max_line_number , 触发⽇志备 份 std::endl; // ⼤于才做备份否则什么⾛不做 Backup(); } } if (_next_handler) // 如果_next_handler被设置就交给下⼀个继续加⼯处理 _next_handler-Execute(info); else std::cout 责任链节点结束,处理完成 std::endl; } void Backup() { std::string newname _filename . std::to_string(time(nullptr)); pid_t id fork(); if (id 0) { chdir(_filepath.c_str()); // 更改进程路径进⼊./tmp/路径下 std::filesystem::rename(_filename, newname); // rename⽐较快也不影 响未来其他继续写⼊的操作因为会重新形成⽂件 std::string tarname newname .tgz; // ⼦进程打包备份 std::cout 打包 : newname 成为: tarname 开始 std::endl; execlp(tar, tar, czf, tarname.c_str(), newname.c_str(), nullptr); // 注意这⾥要以nullptr结尾注意这⾥的坑 std::cout 打包 : newname 成为: tarname 失败 std::endl; exit(1); } waitpid(id, nullptr, 0); std::string tempfile _filepath newname; std::filesystem::remove(tempfile); // 删除⽂件原件只要tar包 } private: int _max_line_number; std::string _filepath; std::string _filename; }; class HandlerEntry { public: HandlerEntry() { // 构建责任链节点对象 _format std::make_sharedHandlerTextFormat(); _save std::make_sharedHandlerTextSaveFile(); _backup std::make_sharedHandlerTextBackupFile(); // 链接责任链 _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_ptrHandlerText _format; std::shared_ptrHandlerText _save; std::shared_ptrHandlerText _backup; };以上代码采用责任链模式实现了一个文本处理流水线。HandlerText是抽象基类持有后继节点和启用开关三个具体处理器分别负责格式化文本、追加写入日志文件、按行数阈值触发备份。HandlerTextFormat给文本加上时间戳和进程号HandlerTextSaveFile把结果写到./tmp/test.logHandlerTextBackupFile检查行数超限后fork子进程用tar打包备份并删除原文件。HandlerEntry负责把三个节点按顺序组装成链并提供统一开关和启动入口。整体思路是请求从链头进入依次经过各节点处理每个节点只关心自己的逻辑处理完交给下一个实现了解耦和可插拔。三、结语本文从System V消息队列的底层机制出发从核心接口和内核数据结构切入最后通过一个责任链模式的文本处理流水线作为实践。消息队列解决的是进程间怎么传的问题责任链解决的是消费端怎么处理的问题。前者让消息异步流动后者让处理逻辑可插拔、可编排。把两者结合消费端就不再是一个巨大的if-else方法而是一条清晰的处理链消息从队列取出后依次经过格式化、落盘、备份等节点每个节点只关心自己的逻辑新增或调整环节只需改动链的组装。用消息队列解决进程间通信的解耦用责任链解决消费端处理逻辑的解耦两者结合就能实现让异步任务从能收发走向可编排。