System V 消息队列-阿里云开发者社区

1、概述

　　消息队列可以认为是一个消息链表，System V 消息队列使用消息队列标识符标识。具有足够特权的任何进程都可以往一个队列放置一个消息，具有足够特权的任何进程都可以从一个给定队列读出一个消息。在某个进程往一个队列写入消息之前，并不需要另外某个进程在该队列上等待消息的到达。System V 消息队列是随内核持续的，只有在内核重起或者显示删除一个消息队列时，该消息队列才会真正被删除。可以将内核中的某个特定的消息队列画为一个消息链表，如下图所示：

对于系统中没个消息队列，内核维护一个msqid_ds的信息结构：

struct msqid_ds
  {
   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 */
};

2、System V 消息队列操作函数

系统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()解除阻塞的条件有三个：

消息队列中有了满足条件的消息；
msqid代表的消息队列被删除；
调用msgrcv（）的进程被信号中断；

调用返回：成功返回读出消息的实际字节数，否则返回-1。

3）int msgsnd(int msqid, struct msgbuf *msgp, int msgsz, int msgflg);
向msgid代表的消息队列发送一个消息，即将发送的消息存储在msgp指向的msgbuf结构中，消息的大小由msgze指定。

对发送消息来说，有意义的msgflg标志为IPC_NOWAIT，指明在消息队列没有足够空间容纳要发送的消息时，msgsnd是否等待。造成msgsnd()等待的条件有两种：

当前消息的大小与当前消息队列中的字节数之和超过了消息队列的总容量；
当前消息队列的消息数（单位"个"）不小于消息队列的总容量（单位"字节数"），此时，虽然消息队列中的消息数目很多，但基本上都只有一个字节。

msgsnd()解除阻塞的条件有三个：

不满足上述两个条件，即消息队列中有容纳该消息的空间；
msqid代表的消息队列被删除；
调用msgsnd（）的进程被信号中断；

调用返回：成功返回0，否则返回-1。

4）int msgctl(int msqid, int cmd, struct msqid_ds *buf);
该系统调用对由msqid标识的消息队列执行cmd操作，共有三种cmd操作：IPC_STAT、IPC_SET 、IPC_RMID。

IPC_STAT：该命令用来获取消息队列信息，返回的信息存贮在buf指向的msqid结构中；
IPC_SET：该命令用来设置消息队列的属性，要设置的属性存储在buf指向的msqid结构中；可设置属性包括：msg_perm.uid、msg_perm.gid、msg_perm.mode以及msg_qbytes，同时，也影响msg_ctime成员。
IPC_RMID：删除msqid标识的消息队列；

调用返回：成功返回0，否则返回-1。

写个程序创建一个消息队列，然后网消息队列中放置一个含有1个字节的消息，发出msgctl的IPC_STAT命令，使用system函数执行ipcs命令，最后使用IPC_RMID命令删除该队列。程序如下：

 
    
  
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

#define MSG_R 0400 /* read permission */
#define MSG_W 0200 /* write permission */

#define SVMSG_MODE (MSG_R | MSG_W | MSG_R >>3 | MSG_R >>6)

typedef unsigned long ulong_t;
struct msgbuf
{
   long mtype;
   char mtext[1];
};

int main()
{
   int msqid;
   struct msqid_ds info;
   struct msgbuf buf;
   if((msqid = msgget(IPC_PRIVATE,SVMSG_MODE|IPC_CREAT)) == -1)
   {
       perror("msgget() error");
       exit(-1);
   }
   buf.mtype = 1;
   buf.mtext[0] = 1;
   if(msgsnd(msqid,&buf,1,0) == -1)
   {
       perror("msgsnd() error");
       exit(-1);
   }
   if(msgctl(msqid,IPC_STAT,&info) == -1)
   {
       perror("msgctl() error");
       exit(-1);
   }
   printf("read-write: %03o,cbytes = %lu,qnum =%lu,qbytes = %lu\n",
          info.msg_perm.mode & 0777,(ulong_t) info.msg_cbytes,
          (ulong_t) info.msg_qnum,(ulong_t) info.msg_qbytes);
   system("ipcs -q");
   if(msgctl(msqid,IPC_RMID,NULL) == -1)
   {
       perror("msgctl() error");
       exit(-1);
   }
   exit(0);
} 
    
  

程序执行结果如下：

System V消息队列是随内核持续的，我们可以编写一组小程序来操作消息队列。程序如下：

创建消息队列msgcreate程序：

 
    
  
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

#define MSG_R 0400 /* read permission */
#define MSG_W 0200 /* write permission */

#define SVMSG_MODE (MSG_R | MSG_W | MSG_R >>3 | MSG_R >>6)

int main(int argc,char *argv[])
{
   int c,oflag,mqid;
   key_t   key;
   oflag = SVMSG_MODE | IPC_CREAT;
   //通过-e命令指定IPC_EXCL标志
   while((c= getopt(argc,argv,"e")) != -1)  
   {
       switch(c)
       {
           case 'e':
               oflag |= IPC_EXCL;
               break;
       }
   }
   if(optind != argc -1)
   {
       printf("usage: msgcreate [-e] <pathname>");
       exit(0);
   }
   //创建key
   if((key = ftok(argv[optind],0)) == -1)
   {
       perror("ftok() error");
       exit(-1);
   }
   //创建或者打开一个消息队列
   if((mqid = msgget(key,oflag)) == -1)
   {
       perror("msgget() error");
       exit(-1);
   }
   exit(0);
} 
    
  

向消息队列放置信息msgsnd程序：

 
    
  
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

#define MSG_W 0200 /* write permission */
//自定义消息结构
struct msgbuf
{
   long mtype;
   char mtext[100];
};

int main(int argc,char *argv[])
{
   int c,oflag,mqid;
   size_t  len;
   long type;
   struct msgbuf *ptr;
   key_t   key;
   if(argc != 4)
   {
       printf("usage: msgsnd <pathname> <#bytes> <type>");
       exit(0);
   }
   len = atoi(argv[2]);
   type = atoi(argv[3]);
   if((key = ftok(argv[1],0)) == -1)
   {
       perror("ftok() error");
       exit(-1);
   }
   if((mqid = msgget(key,MSG_W)) == -1)
   {
       perror("msgget() error");
       exit(-1);
   }
   ptr = calloc(sizeof(long)+len,sizeof(char));
   ptr->mtype = type;
   if(msgsnd(mqid,ptr,len,0) == -1)
   {
       perror("msgsnd() error");
       exit(-1);
   }
   exit(0);
} 
    
  

从消息队列中读出一个消息msgrcv程序：

 
    
  
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

#define MAXMSG  (8192+sizeof(long))
#define MSG_R   0400 /* read permission */

struct msgbuf
{
   long mtype;
   char mtext[100];
};

int main(int argc,char *argv[])
{
   int c,flag,mqid;
   long type;
   ssize_t n;
   struct msgbuf *buff;
   key_t   key;
   type = flag = 0;
   while((c = getopt(argc,argv,"nt:")) != -1)
   {
       switch(c)
       {
       case 'n':
          flag |= IPC_NOWAIT;
          break;
       case 't':
          type = atol(optarg);
          break;
       }
   }
   if(optind != argc -1)
   {
       printf("usage: msgrcv[-n] [-t type] <pathname> ");
       exit(0);
   }
   if((key = ftok(argv[optind],0)) == -1)
   {
       perror("ftok() error");
       exit(-1);
   }
   if((mqid = msgget(key,MSG_R)) == -1)
   {
       perror("msgget() error");
       exit(-1);
   }
   buff = malloc(MAXMSG) ;
   if(buff == NULL)
   {
       perror("malloc() error");
       exit(-1);
   }
   if((n=msgrcv(mqid,buff,MAXMSG,type,flag)) == -1)
   {
       perror("msgsnd() error");
       exit(-1);
   }
   printf("read %d bytes,type =%ld\n",n,buff->mtype);
   exit(0);
} 
    
  

删除一个消息队列msgrmid程序：

 
    
  
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

int main(int argc,char *argv[])
{
   int     mqid;
   key_t   key;
   if(argc != 2)
   {
       printf("usage: msgrmid <pathname>");
       exit(0);
   }
   if((key = ftok(argv[1],0)) == -1)
   {
       perror("ftok() error");
       exit(-1);
   }
   if((mqid = msgget(key,0)) == -1)
   {
       perror("msgget() error");
       exit(-1);
   }
   msgctl(mqid,IPC_RMID,NULL);
   exit(0);
} 
    
  

程序测试结果如下：

写个客户-服务器例子，创建两个消息队列，一个队列用来从客户到服务器的消息，一个队列用于从服务器到客户的消息。主要功能是：客户向服务器发现一条消息，服务接收到并输出接收到的客户消息，然后服务器向客户发送一条消息，客户显示服务器发送的消息。程序如下：

公共头文件svmsg.h：

 
    
  
#ifndef  SVMSG_H
#define  SVMSG_H
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#define MQ_KEY1  1234L       //定义两个KEY
#define MQ_KEY2  2345L
#define MAXMESGDATA    1024
#define MAXMSG  sizeof(long)*2+1024
//消息结构信息
struct msgbuf
{
   long mesg_type;   //消息类型
   char mesg_data[1];  //消息内容
};
#endif 
    
  

客户端程序svmsg_client.c

 
    
  
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "svmsg.h"

void client(int ,int);

int main(int argc,char *argv[])
{
   int readid,writeid;
   //打开消息队列
   //MO_KEY2用于客户从服务器读取
   if((readid = msgget(MQ_KEY2,0)) == -1)   
   {
       perror("msgget() error");
       exit(-1);
   }
   //MO_KEY用于客户向服务器写
   if((writeid = msgget(MQ_KEY1,0))==-1)
   {
       perror("msgget() error");
       exit(-1);
   }
   client(readid,writeid);
   msgctl(readid,IPC_RMID,NULL);  //删除消息队列
   msgctl(writeid,IPC_RMID,NULL);
   exit(0);
}

void client(int readid ,int writeid)
{
   ssize_t n,len;
   struct msgbuf *buff,*ptr;
   char content[MAXMESGDATA];
   long type;
   printf("Enter the msg_type: ");
   scanf("%ld",&type);
   printf("Enter the msg_data: ");
   scanf("%s",content);
   len = strlen(content);
   ptr = calloc(sizeof(long)*2+len,sizeof(char));
   ptr->mesg_type = type;
   strcpy(ptr->mesg_data,content);
   printf("Sent to server.\n");
   if(msgsnd(writeid,ptr,len+sizeof(long),0) == -1)
   {
       perror("msgsnd() error");
       exit(-1);
   }
   printf("Sent to server successfully.\n");
   free(ptr);
   printf("*********************************\n");
   buff = malloc(MAXMSG );
   printf("waiting for server......\n");
   //以阻塞形式读取
   if((n=msgrcv(readid,buff,MAXMSG,0,0)) == -1)
   {
       perror("msgrcv() error");
       exit(-1);
   }
   printf("Received message from server is:\nmsg_type = %ld\nmsg_data = %s\n",
          buff->mesg_type,buff->mesg_data);
   free(buff);
} 
    
  

服务器程序svmsg_server.c

 
    
  
#include "svmsg.h"

#define MSG_R 0400 /* read permission */
#define MSG_W 0200 /* write permission */
#define SVMSG_MODE (MSG_R | MSG_W | MSG_R >>3 | MSG_R >>6)

void server(int ,int);

int main(int argc,char *argv[])
{
   int readid,writeid;
   if((readid = msgget(MQ_KEY1,SVMSG_MODE | IPC_CREAT)) == -1)
   {
       perror("msgget() error");
       exit(-1);
   }
   if((writeid = msgget(MQ_KEY2,SVMSG_MODE | IPC_CREAT)) == -1)
   {
       perror("msgget() error");
       exit(-1);
   }
   server(readid,writeid);
   exit(0);
}

void server(int readid,int writeid)
{
   ssize_t n,len;
   struct msgbuf *buff,*ptr;
   char content[MAXMESGDATA];
   long type;
   buff = malloc(MAXMSG );
   printf("waiting for client......\n");
   if((n=msgrcv(readid,buff,MAXMSG,0,0)) == -1)
   {
       perror("msgrcv() error");
       exit(-1);
   }
   printf("Received message from client is:\nmsg_type = %ld\nmsg_data = %s\n",
          buff->mesg_type,buff->mesg_data);
   free(buff);
   printf("*********************************\n");
   printf("Enter the msg_type: ");
   scanf("%ld",&type);
   printf("Enter the msg_data: ");
   scanf("%s",content);
   len = strlen(content);
   ptr = calloc(sizeof(long)*2+len,sizeof(char));
   ptr->mesg_type = type;
   strcpy(ptr->mesg_data,content);
   printf("Sent to client.\n");
   if(msgsnd(writeid,ptr,len,0) == -1)
   {
       perror("msgsnd() error");
       exit(-1);
   }
   printf("Sent to client successfully.\n");
   free(buff);
} 
    
  

程序测试结果如下：

3、复用消息

　　与一个队列中的每个消息相关联的类型字段提供了两个特性：

（1）类型字段可以用于标识消息，从而允许多个进程在单个队列上复用消息。

（2）类型字段可以用做优先级，允许接收者以不同于先进先出的某个顺序读出各个消息。

System V 消息队列

云消息队列

热门文章

最新文章

相关课程

相关电子书

相关实验场景