linux进程间通信，system V消息队列

1.消息队列共享内存数据传输速度最快但是需要用户自己进行显式同步由于它传递的是一整块内存没有消息的边界概念需要用户自己在协议层面约定而管道天然自带同步但数据是一串没有边界的字节流容易粘包消息队列相比管道和共享内存之间在内核方面提供了共享内存和管道都不具备的能力1.自带同步机制数据的读写由内核自带的队列和阻塞机制保证2.消息有边界一次msgsnd对应一次完整的msgrcv不会出现粘包情况3.可按类型选择性接收每条消息都带有一个类型标识为mtype接收方可以只接受自己需要的类型消息并且多个客户端可以共用一个队列互不干扰2.key_t和ftok消息队列同样通过ftok(pathname, proj_id)生成key_t键值其原理、风险inode依赖、路径重建后key变化、NFS上碰撞概率高与共享内存本身完全一致这里不再重复展开。3.内核数据结构每个消息队列在内核中由一个msqid_ds结构描述struct msqid_ds { struct ipc_perm msg_perm; // 操作权限与所有权 time_t msg_stime; // 最后一次msgsnd的时间 time_t msg_rtime; // 最后一次msgrcv的时间 time_t msg_ctime; // 最后一次修改属性的时间 unsigned long msg_cbytes; // 队列当前已占用的字节数 msgqnum_t msg_qnum; // 队列中当前的消息条数 msglen_t msg_qbytes; // 队列允许的最大字节数受内核参数MSGMNB限制 pid_t msg_lspid; // 最后一次msgsnd的进程PID pid_t msg_lrpid; // 最后一次msgrcv的进程PID };4.用户态消息的格式约定msgsnd / msgrcv 对用户传入的消息缓冲区有一个硬性规定缓冲区第一个成员必须得是long类型作为消息类型mtype且值要大于0后面才是消息体内容。其定义如下struct msgbuf{ long mtype; char mytext[128]; };5.相关系统调用5.1msgget函数创建或获取消息队列int msgget(key_t key,int msgflag);语义与shmget一致IPC_CREAT表示不存在则创建存在则获取IPC_CREAT|IPC_EXCL表示不存在则创建存在则报错需要明确谁是消息队列的创建者。5.2msgsnd函数发送消息int msgsnd(int msqid,const void* msgp,size_t msgsz,int msgflag);msgp指向前面约定格式的消息缓冲区msgsz指消息体的长度不包含mtype这个long类型所占的字节数msgflag常用取值为IPC_NOWAIT队列已满时不阻塞而是直接返回错误不是挂起等待成功返回0失败则返回-15.3msgrcv函数接收消息ssize_t msgrcv(int msqid,void* msgp,size_t msgsz,long msgtyp,int msgflag);msgtyp是消息队列中最独特的参数它决定了接收哪条消息当msgtyp取值为0时接收队列中最早的一条消息无论类型当取值大于0时只接收类型恰好等于msgtyp的第一条消息当取值小于0时接收类型小于等于msgtyp的绝对值中类型值最小的第一条消息可以用来实现简单的优先级语义其中msgtyp0这种精确匹配的能力是实现点对点通信的关键比如约定用发送者的PID作为回复消息的mtype就能让多个客户端共用同一个队列而互不干扰成功返回实际接收到的消息体长度失败则返回-15.4msgctl函数控制消息队列int msgctl(int msqid,int cmd,struct msqid_ds* buf);常用cmdIPC_STAT读取当前状态到buf中IPC_SET按buf更新属性IPC_RMID删除整个队列语义与shmctl一致6.生命周期与清理和共享内存一致其生命周期随内核运行期存在不会随创建其的进程退出而自动释放必须显式调用msgctl(msqid,IPC_RMID,NULL)删除。进程异常崩溃忘记清理统一会导致队列永久残留和共享内存一样可以使用以下命令进行查看。ipcs -q ipcrm -q msqid7.实例基于消息队列的点对点请求应答因为演示的是服务端与客户端一问一答在服务器接收完数据后就直接销毁消息队列了为了避免客户端没有接收到服务端的消息而消息队列就被销毁了。这里稍微用了下sleep函数让服务器进程等了2秒你们可以采用注册退出信号函数等来解决这里就不过多演示了。msg_comm.h#ifndef _MSG_COMM_H_ #define _MSG_COMM_H_ #includestdio.h #include stdlib.h #include string.h #include unistd.h #include sys/types.h #include sys/ipc.h #include sys/msg.h #define PATHNAME . #define PROJ_ID 0x666 #define MSG_TEXT_SIZE 128 //约定所有客户端发送给服务端的请求mytype统一为1 #define REQUEST_TYPE 1 //mtype之外附带发送者pid用于服务端实现只回给一个客户端 struct msgbuf{ long mtype; pid_t pid; char mtext[MSG_TEXT_SIZE]; }; int createmsgqueue(void); int getmsgqueue(void); int destroymsgqueue(int msqid); #endifmsg_comm.c#includemsg_comm.h static int commmsg(int flags) { key_t keyftok(PATHNAME,PROJ_ID); if(key0) { perror(ftok); return -1; } int msqidmsgget(key,flags); if(msqid0) { perror(msgget); return -2; } return msqid; } int createmsgqueue(void) { return commmsg(IPC_CREAT|IPC_EXCL|0666); } int getmsgqueue(void) { return commmsg(0666); } int destroymsgqueue(int msqid) { if(msgctl(msqid,IPC_RMID,NULL)0) { perror(msqid msgctl); return -1; } return 0; }server.c#includemsg_comm.h int main() { int msqidcreatemsgqueue(); if(msqid0) { fprintf(stderr,createmsgqueue fail\n); exit(1); } printf([server] 消息队列就绪,等待请求\n); //定义缓冲区用来存储 struct msgbuf req; //msgtypeREQUEST_TYPE只取“请求”类型的消息其他类型忽略 if(msgrcv(msqid,req,sizeof(req)-sizeof(long),REQUEST_TYPE,0)0) { perror(msgrcv); destroymsgqueue(msqid); exit(1); } printf([server]收到来自客户端pid%d 的请求text%s\n,req.pid,req.mtext); //定义缓冲区来接收数据 struct msgbuf rsp; //用发送者的pid作为mtype实现点对答 rsp.mtypereq.pid; snprintf(rsp.mtext,MSG_TEXT_SIZE,已收到:%s\n,req.mtext); if(msgsnd(msqid,rsp,sizeof(rsp)-sizeof(long),0)0) { perror(msgsend); destroymsgqueue(msqid); exit(1); } printf([server]已收到pid为:%d的回复 %s\n,req.pid,rsp.mtext); //在这里就简化演示实际收发各一次后结束消息队列但在生产环境中通常是常驻循环了 sleep(2); destroymsgqueue(msqid); return 0; }client.c#includemsg_comm.h int main() { int msqidgetmsgqueue(); if(msqid0) { fprintf(stderr,getmsgqueue fail\n); exit(1); } struct msgbuf req; req.mtypeREQUEST_TYPE; req.pidgetpid(); snprintf(req.mtext,MSG_TEXT_SIZE,hello from client); if(msgsnd(msqid,req,sizeof(req)-sizeof(long),0)0) { perror(msgsend); exit(1); } printf([client]请求已发送等待响应中...\n); struct msgbuf rsp; //msgtype等于自己的pid,哪怕消息队列中有其余客户端消息也只会取走发给自己的那条 if(msgrcv(msqid,rsp,sizeof(rsp)-sizeof(long),getpid(),0)0) { perror(msgrcv); exit(1); } printf([client]已收到应答%s\n,rsp.mtext); return 0; }这是输出8.总结消息队列的核心价值是内核自带同步消息自带边界按类型选择性接收基于此消息队列能够做到简单的请求应答式通信但它继承了整个system V IPC家族共同的问题依赖ftok生成key碰撞和路径失效风险、生命周期不随进程退出自动清理、只能在同一台主机的进程间使用。在这里消息队列仅用来理解IPC设计思想的真正适用于高并发分布式等的实际生产环境通常用的是如Kafka这类专门的消息中间件来承担。