linux消息队列进程通信.doc
linux消息队列进程通信 一、消息队列的基本概念消息队列 (也叫做报文队列)是Unix系统V版本中3种进程间通信机制之一。另外两种是信号灯和共享内存。这些IPC机制使用共同的授权方法。只有通过系统调用将标志符传递给核心之后，进程才能存取这些资源。这种系统IPC对象使用的控制方法和文件系统非常类似。使用对象的引用标志符作为资源表中的索引。 消息队列就是一个消息的链表。就是把消息看作一个记录，并且这个记录具有特定的格式以及特定的优先级。对消息队列有写权限的进程可以按照一定的规则添加新消息；对消息队列有读权限的进程则可以从消息队列中读出消息。Linux采用消息队列的方式来实现消息传递。这种消息的发送方式是：发送方不必等待接收方检查它所收到的消息就可以继续工作下去，而接收方如果没有收到消息也不需等待。这种通信机制相对简单，但是应用程序使用起来就需要使用相对复杂的方式来应付了。新的消息总是放在队列的末尾，接收的时候并不总是从头来接收，可以从中间来接收。消息队列是随内核持续的并和进程相关，只有在内核重起或者显示删除一个消息队列时，该消息队列才会真正被删除。因此系统中记录消息队列的数据结构 (struct ipc_ids msg_ids)位于内核中，系统中的所有消息队列都可以在结构msg_ids中中找到访问入口。IPC标识符：每一个I P C目标都有一个唯一的I P C标识符。这里所指的I P C目标是指一个单独的消息队列、一个信号量集或者一个共享的内存段。系统内核使用此标识符在系统内核中指明 I P C目标。IPC 关键字：想要获得唯一的标识符，则必须使用一个 I P C关键字。客户端进程和服务器端进程必须双方都同意此关键字。这是建立一个客户机/服务器框架的第一步。在System V IPC机制中，建立两端联系的路由方法是和I P C关键字直接相关的。通过在应用程序中设置关键字值，每一次使用的关键字都可以是相同的。一般情况下，可以使用f t o k ( )函数为客户端和服务器端产生关键字值。二、ipcs 命令命令ipcs用于读取System V IPC目标的状态。ipcs -q: 只显示消息队列。ipcs -s: 只显示信号量。ipcs -m: 只显示共享内存。ipcs help: 其他的参数。 三、消息队列的主要调用 内核中实现消息传递机制的代码基本上都在文件ipcmsg.c中,消息队列的主要调用有下面4个(1)msgget：调用者提供一个消息队列的键标 (用于表示个消息队列的唯一名字)，当这个消息队列存在的时候， 这个消息调用负责返回这个队列的标识号；如果这个队列不存在，就创建一个消息队列，然后返回这个消息队列的标识号 ，主要由sys_msgget执行。(2)msgsnd：向一个消息队列发送一个消息，主要由sys_msgsnd执行。(3)msgrcv：从一个消息队列中收到一个消息，主要由sys_msgrcv执行。(4)msgctl：在消息队列上执行指定的操作。根据参数的不同和权限的不同，可以执行检索、删除等的操作，主要由sys_msgctl执行。#include <sys/types.h>#include <sys/ipc.h>#include <sys/msg.h>int msgget( key_t msgkey , int flag );取得一个消息队列的ID，如不存在则建立。返回值： 成功时：消息队列的ID 失败时：-1int msgsnd( int msqid , struct msgbuf *msgp , size_t msgsiz , int msgflag );向消息队列送消息返回值： 成功时：0 失败时：-1msqid是消息队列的ID，size_t msgsiz是结构体成员mdata的大小，msgflag与上一章所讲的共享内存的flag起一样的作用，不过，当这个参数为IPC_NOWAIT的时候，如果消息队列已满，则返回错误值。如果不为IPC_NOWAIT，在消息队列已满 的情况下，会一直等到消息队列有空地方的时候再发送。注意这里的这个 struct msgbuf *msgp 。要求的格式如下：struct msgbuf long mtype; char mdata256;long mtype在这里我们用来保存本进程的PID。mdata则是保存要发送的数据。由于mdata的大小不一定（根据实际需要定义），所以这个结构体并没有事先定义好。但是我们定义这个结构体的时候一定要遵循这个规定。你可以改的，只有mdata的大小，和结构体的名称。尽量不要修改结构体成员的名称和类型。实际上，根据mtype，我们还可以有所选择地接受消息。这在下面将会谈到。int msgrcv( int msqid , struct msgbuf *msgp , size_t msgsiz , long msgtyp , int msgflag );从消息队列取得一个消息返回值： 成功时：0 失败时：-1msqid , *msgp , msgsiz不用说了。long msgtyp是结构体msgbuf的mtype成员。msgflag与上述一样。只不过为IPC_NOWAIT的时候，如果消息队列是空的，则等到有消息可读的时候再读。当不为IPC_NOWAIT的时候，如果消息队列是空的，则返回错误值（与字面上理解的有些相反）下面这个链接帮助了我更好地理解了msgrcv 中的 long msgtyp这个参数和struct msgbuf long mtype; char mdata256;里的long mtype这个结构体成员。同样地，为了控制管理消息队列，一样有一个函数msgctl()如下：#include <sys/types.h>#include <sys/ipc.h>#include <sys/msg.h>int msgctl( int msqid , int cmd , struct msqid_ds *buf );返回值： 成功时：0 失败时：-1cmd所指定的值与共享内存部分相同。最后自己写了一个利用消息队列实现进程通信基本思路：我每次将A端发送消息时的消息类型标记为奇数，将B端发送的消息类型为偶数。A端读取消息类型为偶数的消息（也就是说是从B端发送过来的）B端读取消息类型为奇数的消息（也就是说是从A端发送过来的）A.c和B.c 程序基本上是一样的 这个程序还有一些不完善的地方：1.每个端口不能连续输入，必须等对方输入以后才 会显示结果。这里我也不清楚怎么回事。gdb的调试还不太会，错误不太会找。希望高手指教，呵呵2.没有删除消息队列，下次打开时候还会残留以前的消息最后列出参考资料：Linux软件工程师（c语言）实用版linux消息队列操作 分类： linux 2008-03-11 23:32 31857人阅读 评论(10) 收藏 举报 linuxstructbufferfile存储cmd对消息队列的操作无非有下面三种类型：1、 打开或创建消息队列消息队列的内核持续性要求每个消息队列都在系统范围内对应唯一的键值，所以，要获得一个消息队列的描述字，只需提供该消息队列的键值即可；注：消息队列描述字是由在系统范围内唯一的键值生成的，而键值可以看作对应系统内的一条路经。2、 读写操作消息读写操作非常简单，对开发人员来说，每个消息都类似如下的数据结构：struct msgbuflong mtype;char mtext1;mtype成员代表消息类型，从消息队列中读取消息的一个重要依据就是消息的类型；mtext是消息内容，当然长度不一定为1。因此，对于发送消息来说，首先预置一个msgbuf缓冲区并写入消息类型和内容，调用相应的发送函数即可；对读取消息来说，首先分配这样一个msgbuf缓冲区，然后把消息读入该缓冲区即可。3、 获得或设置消息队列属性：消息队列的信息基本上都保存在消息队列头中，因此，可以分配一个类似于消息队列头的结构(struct msqid_ds，见附录 2)，来返回消息队列的属性；同样可以设置该数据结构。消息队列API1、文件名到键值#include <sys/types.h>#include <sys/ipc.h>key_t ftok (char*pathname, char proj)；它返回与路径pathname相对应的一个键值。该函数不直接对消息队列操作，但在调用ipc(MSGGET,)或msgget()来获得消息队列描述字前，往往要调用该函数。典型的调用代码是： key=ftok(path_ptr, 'a'); ipc_id=ipc(MSGGET, (int)key, flags,0,NULL,0); 2.系统V消息队列API系统V消息队列API共有四个，使用时需要包括几个头文件：#include <sys/types.h>#include <sys/ipc.h>#include <sys/msg.h>1）int msgget(key_t key, int msgflg)参数key是一个键值，由ftok获得；msgflg参数是一些标志位。该调用返回与健值key相对应的消息队列描述字。在以下两种情况下，该调用将创建一个新的消息队列：· 如果没有消息队列与健值key相对应，并且msgflg中包含了IPC_CREAT标志位； · key参数为IPC_PRIVATE； 参数msgflg可以为以下：IPC_CREAT、IPC_EXCL、IPC_NOWAIT或三者的或结果。调用返回：成功返回消息队列描述字，否则返回-1。注：参数key设置成常数IPC_PRIVATE并不意味着其他进程不能访问该消息队列，只意味着即将创建新的消息队列。2）int msgrcv(int msqid, struct msgbuf *msgp, int msgsz, long msgtyp, int msgflg);该系统调用从msgid代表的消息队列中读取一个消息，并把消息存储在msgp指向的msgbuf结构中。msqid为消息队列描述字；消息返回后存储在msgp指向的地址，msgsz指定msgbuf的mtext成员的长度（即消息内容的长度），msgtyp为请求读取的消息类型；读消息标志msgflg可以为以下几个常值的或：· IPC_NOWAIT 如果没有满足条件的消息，调用立即返回，此时，errno=ENOMSG · IPC_EXCEPT 与msgtyp>0配合使用，返回队列中第一个类型不为msgtyp的消息 · IPC_NOERROR 如果队列中满足条件的消息内容大于所请求的msgsz字节，则把该消息截断，截断部分将丢失。 msgrcv手册中详细给出了消息类型取不同值时(>0; <0; =0)，调用将返回消息队列中的哪个消息。msgrcv()解除阻塞的条件有三个：1. 消息队列中有了满足条件的消息； 2. msqid代表的消息队列被删除； 3. 调用msgrcv（）的进程被信号中断； 调用返回：成功返回读出消息的实际字节数，否则返回-1。3）int msgsnd(int msqid, struct msgbuf *msgp, int msgsz, int msgflg);向msgid代表的消息队列发送一个消息，即将发送的消息存储在msgp指向的msgbuf结构中，消息的大小由msgze指定。对发送消息来说，有意义的msgflg标志为IPC_NOWAIT，指明在消息队列没有足够空间容纳要发送的消息时，msgsnd是否等待。造成msgsnd()等待的条件有两种：· 当前消息的大小与当前消息队列中的字节数之和超过了消息队列的总容量； · 当前消息队列的消息数（单位"个"）不小于消息队列的总容量（单位"字节数"），此时，虽然消息队列中的消息数目很多，但基本上都只有一个字节。 msgsnd()解除阻塞的条件有三个： 1. 不满足上述两个条件，即消息队列中有容纳该消息的空间； 2. msqid代表的消息队列被删除； 3. 调用msgsnd（）的进程被信号中断； 调用返回：成功返回0，否则返回-1。4）int msgctl(int msqid, int cmd, struct msqid_ds *buf);该系统调用对由msqid标识的消息队列执行cmd操作，共有三种cmd操作：IPC_STAT、IPC_SET 、IPC_RMID。1. IPC_STAT：该命令用来获取消息队列信息，返回的信息存贮在buf指向的msqid结构中； 2. IPC_SET：该命令用来设置消息队列的属性，要设置的属性存储在buf指向的msqid结构中；可设置属性包括：msg_perm.uid、msg_perm.gid、msg_perm.mode以及msg_qbytes，同时，也影响msg_ctime成员。 3. IPC_RMID：删除msqid标识的消息队列； 调用返回：成功返回0，否则返回-1。-/*msgserver.c*/#include <stdlib.h>#include <string.h>#include <errno.h>#include <sys/types.h>#include <sys/ipc.h>#include <sys/msg.h>#include <sys/stat.h>#define MSG_FILE "msgserver.c" #define BUFFER 255 #define PERM S_IRUSR|S_IWUSR /* 服务端创建的消息队列最后没有删除,我们要使用ipcrm命令来删除的 */* ipcrm -q <msqid> */struct msgtype long mtype; char bufferBUFFER+1; ; int main() struct msgtype msg; key_t key; int msgid; if(key=ftok(MSG_FILE,'a')=-1) fprintf(stderr,"Creat Key Error:%s/n", strerror(errno); exit(1); if(msgid=msgget(key, PERM|IPC_CREAT|IPC_EXCL)=-1) fprintf(stderr, "Creat Message Error:%s/n", strerror(errno); exit(1); printf("msqid = %d/n", msgid); while(1) msgrcv(msgid, &msg, sizeof(struct msgtype), 1, 0); fprintf(stderr,"Server Receive:%s/n", msg.buffer); msg.mtype = 2; msgsnd(msgid, &msg, sizeof(struct msgtype), 0); exit(0); /* msgclient.c */#include <stdio.h>#include <stdlib.h>#include <string.h>#include <errno.h>#include <sys/types.h>#include <sys/ipc.h>#include <sys/msg.h>#include <sys/stat.h>#define MSG_FILE "msgserver.c" #define BUFFER 255 #define PERM S_IRUSR|S_IWUSR struct msgtype long mtype; char bufferBUFFER+1; ; int main(int argc, char *argv) struct msgtype msg; key_t key; int msgid; if(argc != 2) fprintf(stderr,"Usage:%s string/n", argv0); exit(1); if(key=ftok(MSG_FILE,'a')=-1) fprintf(stderr,"Creat Key Error:%s/n", strerror(errno); exit(1); if(msgid=msgget(key, PERM)=-1) fprintf(stderr,"Creat Message Error:%s/n", strerror(errno); exit(1); msg.mtype = 1; strncpy(msg.buffer, argv1, BUFFER); msgsnd(msgid, &msg, sizeof(struct msgtype), 0); memset(&msg, '/0', sizeof(struct msgtype); msgrcv(msgid, &msg, sizeof(struct msgtype), 2, 0); fprintf(stderr, "Client receive:%s/n", msg.buffer); exit(0); 今天写了一个小程序，使用了消息队列的msgsnd msgrcv函数，由msgsnd函数循环处理由终端输入的消息，然后把它发送到消息队列，而另一个进程则循环读取消息，进行处理。 这时，问题出现了，每次调用msgrcv函数的时候，它总是第一次调用成功，而第二次返回错误，察看errno=22，打印出来是invalid argument，无效参数。 凭它的说明，可以看出可能是我调用函数的时候参数错误，但为什么第一次能调用成功呢？ 检查了一下，没看出问题。然后google之，发现许多人和我出现了同样的问题，但没有人给出解答。 自己鼓捣了好久，还是没搞定。 然后man 2 msgsnd，一下午不知打了多少遍了，这一次从头到尾一个字一个字的读了下去。 终于发现问题了。 int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg); ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp, int msgflg); man文档里有一句话：The mtext field is an array (or other structure) whose size is specified by msgsz 一直没认真去看，想当然的以为msgsz就是msgp的大小了，原来人家不是，自己自作多情了。 这么一个小问题花了我半个下午，但现在发现总比以后出错要好多了 写出来，给那些第一次使用的朋友们看 把修改后的代码贴出来：struct s_msg long type; char mtext256;/sndint main() int mid; if(mid=msgget(4446,IPC_CREAT|0666)=-1) perr_exit("msgget:"); char bufBUFSIZE; memset(buf,'0',BUFSIZE); s_msg mymsg; while(fgets(buf,BUFSIZE,stdin)!=NULL) if(strlen(buf)<=2)continue; bufstrlen(buf)='0' if(sscanf(buf,"%d%s",&mymsg.type,mymsg.mtext)!=2) perr_exit("Invalid input:"); if(msgsnd(mid,&mymsg,256,IPC_NOWAIT) /msgsiz 为sizeof(mtext),而非sizeof(s_msg) perr_exit("msgsnd:"); memset(buf,'0',BUFSIZE); return 0;/rcvint main(int argc,char *argv) int mid; if(mid=msgget(4446,IPC_CREAT|0666)=-1) perr_exit("msgget:"); s_msg mymsg; while(1) if(msgrcv(mid,&mymsg,256,0,MSG_NOERROR)=-1) /就是这里出错的，记住你了 perr_exit("msgrcv"); if(mymsg.type!=4446) cout<<mymsg.type<<" :"<<mymsg.mtext<<endl; else cout<<"4446 quitn" break; memset(&mymsg,0,sizeof(mymsg); return 0;/*msgsnd 详解*/NAMEmsgsnd - XSI message send operationSYNOPSISXSI #include <sys/msg.h>int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg); DESCRIPTIONThe msgsnd() function operates on XSI message queues (see the Base Definitions volume of IEEE Std 1003.1-2001, Section 3.224, Message Queue). It is unspecified whether this function interoperates with the realtime interprocess communication facilities defined in Realtime.The msgsnd() function shall send a message to the queue associated with the message queue identifier specified by msqid.The application shall ensure that the argument msgp points to a user-defined buffer that contains first a field of type long specifying the type of the message, and then a data portion that holds the data bytes of the message. The structure below is an example of what this user-defined buffer might look like:struct mymsg long mtype; /* Message type. */ char mtext1; /* Message text. */The structure member mtype is a non-zero positive type long that can be used by the receiving process for message selection.The structure member mtext is any text of length msgsz bytes. The argument msgsz can range from 0 to a system-imposed maximum.The argument msgflg specifies the action to be taken if one or more of the following is true:· The number of bytes already on the queue is equal to msg_qbytes; see <sys/msg.h>.· The total number of messages on all queues system-wide is equal to the system-imposed limit.These actions are as follows:· If (msgflg & IPC_NOWAIT) is non-zero, the message shall not be sent and the calling thread shall return immediately.· If (msgflg & IPC_NOWAIT) is 0, the calling thread shall suspend execution until one of the following occurs:o The condition responsible for the suspension no longer exists, in which case the message is sent.o The message queue identifier msqid is removed from the system; when this occurs, errno shall be set equal to EIDRM and -1 shall be returned.o The calling thread receives a signal that is to be caught; in this case the message is not sent and the calling thread resumes execution in the manner prescribed in sigaction().Upon successful completion, the following actions are taken with respect to the data structure associated with msqid; see <sys/msg.h>:· msg_qnum shall be incremented by 1.· msg_lspid shall be set equal to the process ID of the calling process.· msg_stime shall be set equal to the current time.RETURN VALUEUpon successful completion, msgsnd() shall return 0; otherwise, no message shall be sent, msgsnd() shall return -1, and errno shall be set to indicate the error.ERRORSThe msgsnd() function shall fail if:EACCES Operation permission is denied to the calling process; see XSI Interprocess Communication. EAGAIN The message cannot be sent for one of the reasons cited above and (msgflg & IPC_NOWAIT) is non-zero. EIDRM The message queue identifier msqid is removed from the system. EINTR The msgsnd() function was interrupted by a signal. EINVAL The value of msqid is not a valid message queue identifier, or the value of mtype is less than 1; or the value of msgsz is less than 0 or greater than the system-imposed limit. The following sections are informative.EXAMPLESSending a MessageThe following example sends a message to the queue identified by the msqid argument (assuming that value has previously been set). This call specifies that an error should be reported if no message is available. The message size is calculated directly using the sizeof operator.#include <sys/msg.h>.int result;int msqid;struct message long type; char text20; msg;msg.type = 1;strcpy(msg.text, "This is message 1");.result = msgsnd(msqid, (void *) &msg, sizeof(msg.text), IPC_NOWAIT);APPLICATION USAGEThe POSIX Realtime Extension defines alternative interfaces for interprocess communication (IPC). Application developers who need to use IPC should design their applications so that modules using the IPC routines described in XSI Interprocess Communication can be easily modified to use the alternative interfaces.RATIONALENone.FUTURE DIRECTIONSNone.SEE ALSOXSI Interprocess Communication, Realtime, mq_close(), mq_getattr(), mq_notify(), mq_open(), mq_receive(), mq_send(), mq_setattr(), mq_unlink(), msgctl(), msgget(), msgrcv(), sigaction(), t
