15.7 消息队列

消息队列概述 

15.7 消息队列
（1）概述1.msgget: 创建一个新队列或者打开一个现有队列。2.msgsnd：将新消息添加到队列尾部。3.msgrcv：从队列中取数据。注意：并不一定从队列头取数据，也可以按消息的类型或者字段取数据。4.每个消息包含：字段、长度、实际数据字节数。5.为什么需要消息队列?1.管道和FIFO都是字节流的模型,从管道中读出一段数据时,不知道这些数据来自几个消息。2.通过消息队列，无需花费尽力从字节流中解析出完整的消息。3.消息队列中有type字段，消息的读取进程可以通过type字段来选择感兴趣的消息，也可以通过type字段来进行优先级读取，不一定要按照发送的顺序来读取。6.查看系统消息队列支持的最大个数ipcs -q -l	
（2）队列结构struct msqid_ds{struct ipc_perm msg_perm;msgnum_t msg_qnum;msglen_t msg_qbytes;	// 队列中的最大字节数pid_t	 msg_lspid;		// 最后一个调用msgsnd的进程pid_t	 msg_lrpid;		// 最后一个调用msgcv的进程time_t 	msg_stime		// 最后一个调用msgsnd的时间time_t 	msg_rtime		// 最后一个调用msgrcv的时间time_t 	msg_ctime		// 最后一次改变的时间}
（3）创建一个消息队列或者引用一个现有队列int msgget(key_t key, int flag);	// 应该设置XSI IPC对象的读写权限（15.6）key: 整形变量flag: ----------------------------------------------------------------oflag参数				key不存在(ENOENT)	key已经存在无特殊标志				出错,返回-1			成功,获取到已有标识符IPC_CREATE				成功				成功,获取到已有标识符IPC_CREATE|IPCEXCL		成功				出错,返回-1(EEXIST)----------------------------------------------------------------返回值: 	消息队列ID(将key转化为的标识符),出错返回-1注意：应该设置XSI IPC对象的读写权限（15.6）（4）队列控制int msgctl(int msqid, int cmd, struct *buf);cmd:指定对msqid指定的队列要执行的命令IPC_STAT: 取当前队列的msqid_ds结构,并将它存放在buf指定的结构中。IPC_SET:  将某系必要的字段从buf中复制到与队列相关的msqid_ds结构中。IPC_RMID: 从系统中删除该消息队列以及仍在该队列中的所有数据。这种删除立即有效。// IPC_STAT IPC_SET IPC_RMID也可用于信号量和共享内存
（5）发送消息int msgsnd(int msqid, const void *ptr, size_t nbytes, int flag);msqid: msgget返回的IPC标识符IDptr: 有两个成员1.第一个成员type（消息类型）必须为long类型,表示字段。2.第二个参数mtext可以为任意类型,表示消息体。nbytes: 	指定消息体(mtext)中包含的字节数,上线值为MSGMAX。flag: 用于控制msgsnd函数的行为IPC_NOWAIT 如果消息队列满了,立刻失败返回。无: 如果消息队列满了,msgsnd函数会阻塞,直到队列有足够的空间来存放。
（6）接收消息			ssize_t msgrcv(int msqid, void *ptr, size_t nbytes, long type, int flag);前三个参数：与msgsnd一致type:	指定想要哪一种消息	0:  返回队列中的第一个消息>0: 有MSG_EXCEPT标志位：取出消息队列中消息类型为type的第一个消息无MSG_EXCEPT标志位：取出消息队列中消息类型为type的第一个消息<0: 返回消息队列中消息类型值小于等于type绝对值的消息,如果这个消息有多个,就取type值最小的一个。flag:IPC_NOWAIT:1.如果消息队列队列中不存在满足msgtype要求的消息，默认情况阻塞等待，一旦设置IPC_NOWAIT标准，则立即返回失败。MSG_EXCEPT:1.该标志位时Linux特有的2.只有type>0时才有意义,含义是选择消息类型不等于type的第一条消息。注意:1.当消息队列满的情况下, 如果需要取出符合type的消息,会导致进程永远阻塞。所以建议当指定特定的type时,不要阻塞等待。（7）示例1.某个进程向消息队列中发送广告、工作、学习三种消息2.程序1从队列头按顺序取消息3.队列2只取广告消息4.队列3只取广告消息5.队列5只取学习消息// 测试结果看我的博客

测试代码

msg_send.c

// msg_send.c
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <stdbool.h>
#include <unistd.h>
#include <fcntl.h>
#include <assert.h>
#include <sys/wait.h>
#include <sys/acct.h>
#include <errno.h>
#include <sys/times.h>
#include <pthread.h>
#include <signal.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include "mymsg.h"int main()
{key_t key;	// 键int msgid;msgst work;msgst ads;msgst study;int i=0;assert((key_t)-1 != (key = ftok(FTOK_PATH, PRJ_NUM)));assert(-1 != (msgid = msgget(key, IPC_CREAT | S_IRUSR| S_IWUSR)));printf("msgid=%d\n", msgid);work.type	= WORK;ads.type	= ADS;study.type	= STUDY;while (1)	// 发送消息{i++;sprintf(work.msg.msgdata, "work msg: %d", i);sprintf(ads.msg.msgdata, "ads msg: %d", i);sprintf(study.msg.msgdata, "study msg: %d", i);work.msg.msgdata[DATALEN] = '\0';ads.msg.msgdata[DATALEN] = '\0';study.msg.msgdata[DATALEN] = '\0';printf("snd %s\n", work.msg.msgdata);printf("snd %s\n", ads.msg.msgdata);printf("snd %s\n", study.msg.msgdata);assert(-1 != msgsnd(msgid, &work, SENDLEN, NOFLAG));assert(-1 != msgsnd(msgid, &ads, SENDLEN, NOFLAG));assert(-1 != msgsnd(msgid, &study, SENDLEN, NOFLAG));printf("snd OK\n");sleep(1);}return 0;
}

mymsg.h 

// mymsg.h
#ifndef MYMSG_H
#define MYMSG_H#define DATALEN 255
#define SENDLEN sizeof(msg_t)
#define RECVLEN SENDLENtypedef enum msgtype{ADS	 = 1L,WORK = 2L,STUDY = 3L
}msgtype;#define ODERRECV  0
#define RECVADS   ADS
#define RECVWORK  WORK
#define RECVSTUDY STUDY
#define NOFLAG    0typedef struct{char msgdata[DATALEN+1];int res;	// 保留
}msg_t;typedef struct msgst{long type;	// 消息类型msg_t msg;	// 可以为任意结构类型或者变量类型
}msgst;#define FTOK_PATH "/data1/zhouhouping/nfs/testCode/chapter15/msg/create_key"
#define PRJ_NUM	12346#endif

oder_recv.c 

oder_recv.c#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <stdbool.h>
#include <unistd.h>
#include <fcntl.h>
#include <assert.h>
#include <sys/wait.h>
#include <sys/acct.h>
#include <errno.h>
#include <sys/times.h>
#include <pthread.h>
#include <signal.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include "mymsg.h"/* 部分执行结果:msgid=163845recv work msg: 18recv ads msg: 18recv study msg: 18recv work msg: 19recv ads msg: 19recv study msg: 19recv work msg: 20recv ads msg: 20recv study msg: 20recv work msg: 21
*/
int main()
{key_t key;	// 键int msgid;msgst newmsg;int i=0;assert((key_t)-1 != (key = ftok(FTOK_PATH, PRJ_NUM)));assert(-1 != (msgid = msgget(key, NOFLAG | S_IRUSR| S_IWUSR))); // 不存在该消息队列就返回错误printf("msgid=%d\n", msgid);while(1){assert(-1 != msgrcv(msgid, &newmsg, RECVLEN, ODERRECV, NOFLAG));	// 没有消息会阻塞printf("recv %s\n", newmsg.msg.msgdata);}return 0;
}

recv_ads.c

// recv_ads.c#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <stdbool.h>
#include <unistd.h>
#include <fcntl.h>
#include <assert.h>
#include <sys/wait.h>
#include <sys/acct.h>
#include <errno.h>
#include <sys/times.h>
#include <pthread.h>
#include <signal.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include "mymsg.h"/* 部分执行结果:recv ads msg: 181recv ads msg: 182recv ads msg: 183recv ads msg: 184recv ads msg: 185recv ads msg: 186recv ads msg: 187recv ads msg: 188
*/
int main()
{key_t key;	// 键int msgid;msgst newmsg;int i=0;assert((key_t)-1 != (key = ftok(FTOK_PATH, PRJ_NUM)));assert(-1 != (msgid = msgget(key, NOFLAG | S_IRUSR| S_IWUSR))); // 不存在该消息队列就返回错误printf("msgid=%d\n", msgid);while(1){assert(-1 != msgrcv(msgid, &newmsg, RECVLEN, RECVADS, NOFLAG));	// 没有消息会阻塞printf("recv %s\n", newmsg.msg.msgdata);}return 0;
}

// recv_study.c

//recv_study.c#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <stdbool.h>
#include <unistd.h>
#include <fcntl.h>
#include <assert.h>
#include <sys/wait.h>
#include <sys/acct.h>
#include <errno.h>
#include <sys/times.h>
#include <pthread.h>
#include <signal.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include "mymsg.h"/* 部分执行结果:msgid=163845recv study msg: 35recv study msg: 36recv study msg: 37recv study msg: 38recv study msg: 40recv study msg: 41recv study msg: 42
*/
int main()
{key_t key;	// 键int msgid;msgst newmsg;int i=0;assert((key_t)-1 != (key = ftok(FTOK_PATH, PRJ_NUM)));assert(-1 != (msgid = msgget(key, NOFLAG | S_IRUSR| S_IWUSR))); // 不存在该消息队列就返回错误printf("msgid=%d\n", msgid);while(1){assert(-1 != msgrcv(msgid, &newmsg, RECVLEN, RECVSTUDY, NOFLAG));	// 没有消息会阻塞printf("recv %s\n", newmsg.msg.msgdata);}return 0;
}

// recv_work.c

// recv_work.c
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <stdbool.h>
#include <unistd.h>
#include <fcntl.h>
#include <assert.h>
#include <sys/wait.h>
#include <sys/acct.h>
#include <errno.h>
#include <sys/times.h>
#include <pthread.h>
#include <signal.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include "mymsg.h"/* 部分执行结果:msgid=163845recv work msg: 20recv work msg: 22recv work msg: 24recv work msg: 26recv work msg: 27recv work msg: 28recv work msg: 29recv work msg: 30recv work msg: 31
*/
int main()
{key_t key;	// 键int msgid;msgst newmsg;int i=0;assert((key_t)-1 != (key = ftok(FTOK_PATH, PRJ_NUM)));assert(-1 != (msgid = msgget(key, NOFLAG | S_IRUSR| S_IWUSR))); // 不存在该消息队列就返回错误printf("msgid=%d\n", msgid);while(1){assert(-1 != msgrcv(msgid, &newmsg, RECVLEN, RECVWORK, NOFLAG));	// 没有消息会阻塞printf("recv %s\n", newmsg.msg.msgdata);}return 0;
}