Linux 进程间通信 --消息队列

Posted 2020-10-11

一、消息队列

　　消息队列提供了一种从一个进程向另一个进程发送一个数据块的方法。

　　每个数据块都被认为含有一个类型，接收进程可以独立地接收含有不同类型的数据结构。我们可以通过发送消息来避免命名管道的同步和阻塞问题。

　　但是消息队列与命名管道一样，每个数据块都有一个最大长度的限制。

　　消息队列是系统内核地址空间中的一个内部的链表。消息可以按照顺序发送到队列中，也可以以几种不同的方式从队列中读取。每一个消息队列用一个唯一的IPC标识符表示。

二、

　　在<sys/msg.h>中，消息队列的数据结构是这样定义的：

struct msgbuf｛  
     long mtype;         /*type of message, must＞0*/  
    char mtext[1];      /*message text*/  
｝；  

 在数据结构msgbuf中共有两个元素：

     mtype指消息的类型，它由一个整数来代表，并且它只能是大于0的整数。

     mtext是消息数据本身。

 mtext字段不但可以存储字符，还可以存储任何其他的数据类型。此字段可以说是完全任意的，因为程序员自己可以重新定义此数据结构。请看下面重新定义的例子：

struct my_msgbuf{  
    long mtype;           /*Message type*/  
    char request_id;      /*Request identifier*/  
    struct client info;   /*Client information structure*/  
};  

 这里的消息类型字段和前面的一样，但数据结构的其余部分则由其他的两个字段所代替，而其中的一个还是另外一个结构。这就体现了消息队列的灵活之处。内核本身并不对消息结构中的数据做任何翻译。你可以在其中发送任何信息，但存在一个内部给定的消息大小的限制。在Linux系统中，消息的最大的长度是4056个字节，其中包括mtype，它占用4个字节的长度。

三、创建消息队列  msgget()

int msgget(key_t key, int msgflg);  

 系统调用msgget()中的第一个参数是消息队列关键字值，可以由ftok()获得。第二个参数msgflg是一些标志，包括：

   IPC_CREAT：如果内核中没有此队列，则创建它。

   IPC_EXCL：当和IPC_CREAT一起使用时，如果队列已经存在，则返回错误。

 当msgget()执行成功时，返回消息队列的标识符，否则返回-1，通过errno和perror()函数查看错误信息.

下面是一个打开和创建一个消息队列的例子，函数返回消息队列的标识符

 

int open_ queue(key_ t keyval)  
{  
  int qid;  
  if((qid = msgget(keyval，IPC_CREAT|0660)) == -1){  
    perror(”msgget”);  
    return(-1);  
  ｝  
  return(qid);  
｝ 

发送和接收消息 msgsnd()

当得到了消息队列标识符，就可以在队列上执行发送或者接收消息了。msgsnd()系统调用用于向队列发送一条消息，其函数原型是：

int msgsnd(int msqid, struct msgbuf *msgp, sizet msgsz, int msgflg); 

　　第一个参数是消息队列标识符。

　　第二个参数msgp，是指向消息缓冲区的指针。

　　参数msgsz指定了消息的字节大小，但不包括消息类型的长度(4个字节)。

　　参数msgflg可以设置为：

  　　　　　　 0：此时为忽略此参数，如果消息队列已满，调用进程将会挂起，直到消息可以写入到队列中。

 　　　　　　IPC_NOWAIT：如果消息队列己满，那么此消息则不会写入到消息队列中，控制将返回到调用进程中。消息队列写入成功时，函数返回0，否则返回-1。

int send_message(int qid, struct mymsgbuf *qbuf)  
{  
  int result, length;  
  /*The length is essentially the size of the structure minus sizeof(mtype)*/  
  length = sizeof(struct mymsgbuf) - sizeof(long);  
  if((result = msgsnd(qid, qbuf, length, O)) == -1){  
     perror(”msgsnd”);  
     return(-1);  
  ｝  
  return(result);  
｝ 

msgrcv()系统调用用于从消息队列读取一条消息，其函数原型是：

ssize_t msgrcv(int msqid, struct msgbuf *msgp, size_t msgsz, long msgtype, int msgflg); 

　　　　第一个参数是消息队列的标识符。

　　　　第二个参数代表要存储消息的缓冲区的地址。

　　　　第三参数是消息缓冲区的长度，不包括mtype的长度，它可以按照如下的方法计算：

　　　　　　msgsz=sizeof(struct mymsgbuf)-sizeof(long)；

　　　　第四个参数是要从消息队列中读取的消息的类型。

　　　　如果msgtype=0,接收消息队列的第一个消息。大于0接收队列中消息类型等于这个值的第一个消息。小于0接收消息队列中小于或者等于msgtype绝对值的所有消息中的最小一个消息。一般为0。

　　　　第五个参数msgflg取值为：

  　　　　　　　　0：从队列中取出最长时间的一条消息。

 　　　　　　　　IPC_NOWAIT：当队列没有消息时，调用会立即返回ENOMSG错误。否则，调用进程将会挂起，直到队列中的一条消息满足msgrcv()的参数要求。

   　　　　当函数成功时，返回写入缓冲区的数据大小，否则返回-1。

  下面是一个接收消息的例子：

read_message(int qid, long type, struct mymsgbuf *qbuf)  
{      
int result, length;  
/* The length is essentially the size of the structure minus sizeof(mtype)*/  
length = sizeof(struct mymsgbuf) - sizeof(long);  
  if((result = msgrcv(qid, qbuf, length, type, O))==-l){  
  perror(¨msgrcv¨);  
  return(-1);  
}  
return(result)；  
｝  

消息队列的控制

消息队列标识符的属性被记录在一个msgid_ds结构体：

struct msqid_ds{  
  struct ipc_perm msg_perm;      /*所有者和权限*/  
  time_t          msg_stime;     /*最后一次向队列发送消息的时间*/  
  time_t          msg_rtime;     /*最后一次从队列接收消息的时间*/  
  time_t          msg_ctime;     /*队列最后一次改动的时问*/  
  unsigned long   __msg_cbytes;  /*当前队列所有消息的总长度*/  
  msgqnum_t       msg_qnum;      /*当前队列中的消息数量*/  
  msglen_t        msg_qbytes;    /*消息队列的最大消息总长度*/  
  pid_t           msg_lspid;     /*最一次给队列发送消息的进程PID*/  
  pid_t           msg_lrpid;     /*最后一次从队列接收消息的进程PID*/  
}  

通过msgctl()可以对消息队列进行控制或者一些属性的修改，其函数原型为：

int msgctl(int msqid, int cmd, struct msqid ds *buf; 

　　第一个参数是消息队列的标识符，第二个参数cmd指定了操作，下面是几个常用的操作：

  　　　　IPC_STAT：读取消息队列的数据结构msqid_ds，并将其存储在buf指定的地址中。

  　　　　IPC_SET：设置消息队列的数据结构msqid_ds中的ipc_perm、msg_qbytes、msg_ctime元素的值。这个值取自buf参数。   

  　　　　IPC_RMID：从系统内核中移走消息队列。

　　比如下面是一个删除消息队列的例子。

int remove_queue(int qid){  
  if(msgctl(qid, IPC_RMID, 0)==-1){  
    perror(”msgctl”);  
    return(-1);  
｝  
  return(O);  
｝

测试：

 

#include <sys/types.h>
#include <sys/msg.h>
#include <unistd.h>

struct msg_buf
    {
        int mtype;
        char data[255];
    };
 
int main()
{
        key_t key;
        int msgid;
        int ret;
        struct msg_buf msgbuf;
 
        key=ftok("/tmp/2",‘a‘);
        printf("key =[%x]\n",key);
        msgid=msgget(key,IPC_CREAT|0666); /*通过文件对应*/

        if(msgid==-1)
        {
                printf("create error\n");
                return -1;
        }
 
        msgbuf.mtype = getpid();
        strcpy(msgbuf.data,"test haha");
        ret=msgsnd(msgid,&msgbuf,sizeof(msgbuf.data),IPC_NOWAIT);
        if(ret==-1)
        {
                printf("send message err\n");
                return -1;
        }
 
        memset(&msgbuf,0,sizeof(msgbuf));
        ret=msgrcv(msgid,&msgbuf,sizeof(msgbuf.data),getpid(),IPC_NOWAIT);
        if(ret==-1)
        {
                printf("recv message err\n");
                return -1;
        }
        printf("recv msg =[%s]\n",msgbuf.data);
 
}