进程间通信——消息队列

原文地址：http://hi.baidu.com/monalisa88188/blog/item/f5c52122e42ff84dac34de74.html
一、消息队列的基本概念
    消息队列是一个存放在内核中的消息链表，每个消息队列由消息队列标识符标识。与管道不同的是消息队列存放在内核中，只有在内核重启（即操作系统重启）或者显示地删除一个消息队列时，该消息队列才会被真正删除。
    操作消息队列时，需要用到一些数据结构，熟悉这些数据结构是掌握消息队列的关键。下面介绍几个重要的数据结构。
    1、消息缓冲结构
    向消息队列发送消息时，必须组成合理的数据结构。Linux系统定义了一个模板数据结构msgbuf：

#include
struct msgbuf{
long mtype;
char mtext[1];
};

#include 
struct msgbuf{
long mtype;
char mtext[1];
};

    结构体中的mtype字段代表消息类型。给消息指定类型，可以使得消息在一个队列中重复使用。mtext字段指消息内容。
    注意：mtext虽然定义为char类型，并不代表消息只能是一个字符，消息内容可以为任意类型，由用户根据需要定义。如下面就是用户定义的一个消息结构：

struct myMsgbuf{
long mtype;
struct student stu;
};

struct myMsgbuf{
long mtype;
struct student stu;
};

    消息队列中的消息的大小是受限制的，由中的宏MSGMAX给出消息的最大长度，在实际应用中要注意这个限制。
    2、msgqid_ds内核数据结构
    Linux内核中，每个消息队列都维护一个结构体msqid_ds，此结构体保存着消息队列当前的状态信息。该结构定义在头文件linux/msg.h中，具体定义如下：

struct msqid_ds{
struct_ipc_perm      msg_perm;
struct_msg              *msg_first;
struct_msg              *msg_last;
__kernel_t time_t      msg_stime;
__kernel_t time_t      msg_rtime;
__kernel_t time_t      msg_ctime;
unsigned long          msg_lcbytes;
unsigned long          msg_lqbytes;
unsigned short        msg_cbytes;
unsigned short        msg_qnum;
unsigned short        msg_qbytes;
__kernel_ipc_pid_t    msg_lspid;
__kernel_ipc_pid_t    msg_lrpid;
};

struct msqid_ds{
struct_ipc_perm      msg_perm;
struct_msg              *msg_first;
struct_msg              *msg_last;
__kernel_t time_t      msg_stime;
__kernel_t time_t      msg_rtime;
__kernel_t time_t      msg_ctime;
unsigned long          msg_lcbytes;
unsigned long          msg_lqbytes;
unsigned short        msg_cbytes;
unsigned short        msg_qnum;
unsigned short        msg_qbytes;
__kernel_ipc_pid_t    msg_lspid;
__kernel_ipc_pid_t    msg_lrpid;
};

    各字段的含义如下：
    msg_perm:是一个ipc_perm（定义在头文件linux/ipc.h）的结构，保存了消息队列的存取权限，以及队列的用户ID、组ID等信息。
    msg_first：指向队列中的第一条消息
    msg_last：指向队列中的最后一条消息
    msg_stime：向消息队列发送最后一条信息的时间
    msg_rtime：从消息队列取最后一条信息的时间
    msg_ctime：最后一次变更消息队列的时间
    msg_cbytes：消息队列中所有消息占的字节数
    msg_qnum：消息队列中消息的数目
    msg_qbytes：消息队列的最大字节数
    msg_lspid：向消息队列发送最后一条消息的进程ID
    msg_lrpid：从消息队列读取最后一条信息的进程ID
    3、ipc_perm：内核数据结构
    结构体ipc_perm保存着消息队列的一些重要的信息，比如消息队列关联的键值，消息队列的用户ID、组ID等，它定义在头文件linux/ipc.h中：

struct ipc_perm{
__kernel_key_t      key;
__kernel_uid_t       uid;
__kernel_gid_t       gid;
__kernel_uid_t       cuid;
__kernel_gid_t       cgid;
__kernel_mode_t   mode;
unsigned_short     seg;
};

struct ipc_perm{
__kernel_key_t      key;
__kernel_uid_t       uid;
__kernel_gid_t       gid;
__kernel_uid_t       cuid;
__kernel_gid_t       cgid;
__kernel_mode_t   mode;
unsigned_short     seg;
};

    几个主要字段的含义如下：
    key：创建消息队列用到的键值key
    uid：消息队列的用户ID
    gid：消息队列的组ID
    cuid：创建消息队列的进程用户ID
    cgid：创建消息队列的进程组ID

二、消息队列的创建与读写
    1、创建消息队列
    消息队列是随着内核的存在而存在的，每个消息队列在系统范围内对应惟一的键值。要获得一个消息队列的描述符，只需提供该消息队列的键值即可，该键值通常由函数ftok返回。该函数定义在头文件sys/ipc.h中：
#include
#include
key_t ftok(const char *pathname,int proj_id);
    ftok函数根据pathname和proj_id这两个参数生成惟一的键值。该函数执行成功会返回一个键值，失败返回－1。一个获取键值的例子：
 

#include
#include
#include

int main(void)
{
int i;
for(i = 1;i <= 5;i++)
    printf("key[%d] = %ul \n",i,ftok(".",i));
exit(0);
}

#include 
#include 
#include 

int main(void)
{
int i;
for(i = 1;i <= 5;i++)
    printf("key[%d] = %ul \n",i,ftok(".",i));
exit(0);
}

    注意：参数pathname在系统中一定要存在且进程有权访问，参数proj_id的取值范围为1～125。
    ftok()返回的键值可以提供给函数msgget。msgget()根据这个键值创建一个新的消息队列或者访问一个已存在的消息队列。msgget定义在头文件sys/msg.h中：
int msgget(key_t key,int msgflg);
    msgget的参数key即为ftok函数的返回值。msgflg是一个标志参数。以下是msgflg的可能取值。
    IPC_CREATE：如果内核中不存在键值与key相等的消息队列，则新建一个消除队列：如果存在这样的消息队列，返回该消息队列的描述符。
    IPC_EXCL：和IPC_CREATE一起使用，如果对应键值的消息队列已经存在，则出错，返回－1。
    注意：IPC_EXCL单独使用是没有任何意义的。
    该函数如果调用成功返回一个消息队列的描述符，否则返回－1。
    2、写消息队列
    创建一个消息队列后，就可以对消息队列进行读写了。函数msgsnd用于向消息队列发送（写）数据。该函数定义在头文件sys/msg.h中：
int msgsnd(int msgid,struct msgbuf *msgp,size_t msgsz,int msgflg);
    msgsnd各参数含义如下：
    msgid：函数向msgid标识的消息队列发送一个消息。
    msgp：msgp指向发送的消息。
    msgsz：要发送的消息的大小，不包含消息类型占用的4个字节。
    msgflg：操作标志位。可以设置为0或者IPC_NOWAIT。如果msgflg为0，则当消息队列已满的时候，msgsnd将会阻塞，直到消息可以写进消息队列；如果msgflg为IPC_NOWAIT，当消息队列已满的时候，msgsnd函数将不等待立即返回。
    msgsnd函数成功返回0，失败返回－1。常见错误码有：EAGAIN，说明消息队列已满；EIDRM，说明消息队列已被删除；EACCESS，说明无权访问消息队列。
/*写消息*/

#include
#include
#include
#include

#define BUF_SIZE  256
#define PROJ_ID   32
#define PATH_NAME "."
int main(void){
struct mymsgbuf{
  long msgtype;
  char ctrlstring[BUF_SIZE];

} msgbuffer;
int qid;
int msglen;
key_t msgkey;

/*get key value*/
if((msgkey=ftok(PATH_NAME,PROJ_ID))==-1)
{
  perror("ftok error!\n");
  exit(1);
}

/*creat message queue*/
if((qid=msgget(msgkey,IPC_CREAT|0660))==-1)
{
  perror("msgget error!\n");

}

/*fill message struct, and send the message*/

msgbuffer.msgtype=3;
strcpy(msgbuffer.ctrlstring,"Hello,message queue");
msglen=sizeof(struct mymsgbuf)- sizeof(long);

if(msgsnd(qid,&msgbuffer,msglen,0)==-1)
{
  perror("msgget error!\n");
  exit(1);
}

exit(0);
}

#include 
#include 
#include 
#include 

#define BUF_SIZE  256
#define PROJ_ID   32
#define PATH_NAME "."
int main(void){
struct mymsgbuf{
  long msgtype;
  char ctrlstring[BUF_SIZE];

} msgbuffer;
int qid;
int msglen;
key_t msgkey;

/*get key value*/
if((msgkey=ftok(PATH_NAME,PROJ_ID))==-1)
{
  perror("ftok error!\n");
  exit(1);
}

/*creat message queue*/
if((qid=msgget(msgkey,IPC_CREAT|0660))==-1)
{
  perror("msgget error!\n");

}

/*fill message struct, and send the message*/

msgbuffer.msgtype=3;
strcpy(msgbuffer.ctrlstring,"Hello,message queue");
msglen=sizeof(struct mymsgbuf)- sizeof(long);

if(msgsnd(qid,&msgbuffer,msglen,0)==-1)
{
  perror("msgget error!\n");
  exit(1);
}

exit(0);
}

    3、读消息队列

    消息队列中放入数据后，其他进程就可以读取其中的消息了。读取消息的系统调用为msgrcv()，该函数定义在头文件sys/msg.h中，其原型如下：
int msgrcv(int msqid,struct msgbuf *msgp,size_t msgsz,long int msgtyp,int msgflg);
    该函数有5个参数，含义如下：
    msqid：消息队列描述符。
    msgp：读取的消息存储到msgp指向的消息结构中。
    msgsz：消息缓冲区的大小。
    msgtyp：为请求读取的消息类型。
    msgflg：操作标志位。msgflg可以为IPC_NOWAIT,IPC_EXCEPT,IPC_NOERROR3个常量。这些值的意义分别为：
IPC_NOWAIT，如果没有满足条件的消息，调用立即返回，此时错误代码为ENOMSG；
IPC_EXCEPT，与msgtyp配合使用，返回队列中第一个类型不为msgtyp的消息；
IPC_NOERROR，如果队列中满足条件的消息内容大于所请求的msgsz字节，则把该消息截断，截断部分将被丢弃。
    调用msgrcv函数的时候，成功会返回消息的实际字节数，否则返回－1。常见的错误码有：E2BIG，表示消息的长度大于msgsz；EIDRM，表示消息队列已被删除；EINVAL，说明msqid无效或msgsz小于0。

/*读消息*/

#include
#include
#include
#include

#define BUF_SIZE  256
#define PROJ_ID   32
#define PATH_NAME "."
int main(void)
{

struct mymsgbuf{
 long msgtype;
 char ctrlstring[BUF_SIZE];

} msgbuffer;
int qid;
int msglen;
key_t msgkey;

if((msgkey=ftok(PATH_NAME,PROJ_ID))==-1)
{
  perror("ftok error!\n");
  exit(1);
}


if((qid=msgget(msgkey,IPC_CREAT|0660))==-1)
{
  perror("msgget error!\n");
  exit(1);
}

msglen=sizeof(struct mymsgbuf)- sizeof(long);
if(msgrcv(qid,&msgbuffer,msglen,3.0)==-1)
{
  perror("msgrcv error!\n");
  exit(1);
}

printf("Get message: %s\n", msgbuffer.ctrlstring);
exit(0);
}

#include 
#include 
#include 
#include 

#define BUF_SIZE  256
#define PROJ_ID   32
#define PATH_NAME "."
int main(void)
{
 
struct mymsgbuf{
 long msgtype;
 char ctrlstring[BUF_SIZE];
 
} msgbuffer;
int qid;
int msglen;
key_t msgkey;
 
if((msgkey=ftok(PATH_NAME,PROJ_ID))==-1)
{
  perror("ftok error!\n");
  exit(1);
}

 
if((qid=msgget(msgkey,IPC_CREAT|0660))==-1)
{
  perror("msgget error!\n");
  exit(1);
}

msglen=sizeof(struct mymsgbuf)- sizeof(long);
if(msgrcv(qid,&msgbuffer,msglen,3.0)==-1)
{
  perror("msgrcv error!\n");
  exit(1);
}

printf("Get message: %s\n", msgbuffer.ctrlstring);
exit(0);
}

三、获取和设置消息队列的属性    消息队列的属性保存在系统维护的数据结构msqid_ds中，用户可以通过函数msgctl获取或设置消息队列的属性。msgctl定义在头文件sys/msg.h中，如下：
int msgctl(int msqid,int cmd,struct msqid_ds *buf);
    msgctl系统调用对msqid标识的消息队列执行cmd操作，系统定义了3种cmd操作：IPC_STAT,IPC_SET,IPC_RMID，它们的意义如下：
    IPC_STAT：该命令用来获取消息队列对应的msqid_ds数据结构，并将其保存到buf指向的地址空间。
    IPC_SET：该命令用来设置消息队列的属性，要设置的属性存储在buf中，可设置的属性包括：msg_perm.uid,msg_perm.gid,msg_perm.mode以及msg_qbytes。
    IPC_RMID：从内核中删除msqid标识的消息队列。

#include
#include
#include
#include

#define BUF_SIZE  256
#define PROJ_ID   32
#define PATH_NAME "."

void getmsgattr(int msgid,struct msqid_ds msq_info);
int main(void)
{
/*  */
struct mymsgbuf{
 long msgtype;
 char ctrlstring[BUF_SIZE];

} msgbuffer;
int qid;
int msglen;
key_t msgkey;
struct msqid_ds msg_attr;
/*获取键值*/
if((msgkey=ftok(PATH_NAME,PROJ_ID))==-1)
{
  perror("ftok error!\n");
  exit(1);

}

/*获取消息队列标识符*/
if((qid=msgget(msgkey,IPC_CREAT|0660))==-1)
{
  perror("msgget error!\n");
  exit(1);

}
getmsgattr(qid,msg_attr); /*输出消息队列的属性*/

/*发送一条消息到消息队列*/
msgbuffer.msgtype=2;
strcpy(msgbuffer.ctrlstring,"Another message");
msglen=sizeof(struct mymsgbuf)- sizeof(long);
if(msgsnd(qid,&msgbuffer,msglen,0)==-1)
{
  perror("msgget error!\n");
  exit(1);
}

getmsgattr(qid,msg_attr); /*再输出消息队列的属性*/

/*设置消息队列的属性*/
msg_attr.msg_perm.uid = 8;
msg_attr.msg_perm.gid = 8;
if (msgctl(qid,IPC_SET,&msg_attr) == -1)
{
      perror("msg set error!\n");
      exit(1);
}
getmsgattr(qid,msg_attr);/*修改后再观察其属性*/
if (msgctl(qid,IPC_RMID,NULL) == -1)
{
      perror("delete msg error!\n");
      exit(1);
}
getmsgattr(qid,msg_attr);/*删除后再观察其属性*/
}

void getmsgattr(int msgid,struct msqid_ds msg_info)
{
    if (msgctl(msgid,IPC_STAT,&msg_info) == -1)
    {
      perror("msgctl error!\n");
      return;
    }

    printf("***information of message queue %d***\n",msgid);
    printf("last msgsnd to msq time is %s\n",ctime(&(msg_info.msg_stime)));
    printf("last msgrcv time from msg is %s\n",ctime(&(msg_info.msg_rtime)));
    printf("last change msg time is %s\n",ctime(&(msg_info.msg_ctime)));
    printf("current number of bytes on queue is %d\n",msg_info.msg_cbytes);
    printf("number of messages in queue is %d\n",msg_info.msg_qnum);
    printf("max number of bytes on queue is %d\n",msg_info.msg_qbytes);
    printf("pid of last msgsnd is %d\n",msg_info.msg_lspid);
    printf("pid of last msgrcv is %d\n",msg_info.msg_lrpid);
    printf("msg uid is %d\n",msg_info.msg_perm.uid);
    printf("msg gid is %d\n",msg_info.msg_perm.gid);
    printf("*******information end!**************\n",msgid);
}

#include 
#include 
#include 
#include 

#define BUF_SIZE  256
#define PROJ_ID   32
#define PATH_NAME "."

void getmsgattr(int msgid,struct msqid_ds msq_info);
int main(void)
{
/*  */
struct mymsgbuf{
 long msgtype;
 char ctrlstring[BUF_SIZE];
 
} msgbuffer;
int qid;
int msglen;
key_t msgkey;
struct msqid_ds msg_attr;
/*获取键值*/
if((msgkey=ftok(PATH_NAME,PROJ_ID))==-1)
{
  perror("ftok error!\n");
  exit(1);

}

/*获取消息队列标识符*/
if((qid=msgget(msgkey,IPC_CREAT|0660))==-1)
{
  perror("msgget error!\n");
  exit(1);

}
getmsgattr(qid,msg_attr); /*输出消息队列的属性*/

/*发送一条消息到消息队列*/
msgbuffer.msgtype=2;
strcpy(msgbuffer.ctrlstring,"Another message");
msglen=sizeof(struct mymsgbuf)- sizeof(long);
if(msgsnd(qid,&msgbuffer,msglen,0)==-1)
{
  perror("msgget error!\n");
  exit(1);
}

getmsgattr(qid,msg_attr); /*再输出消息队列的属性*/

/*设置消息队列的属性*/
msg_attr.msg_perm.uid = 8;
msg_attr.msg_perm.gid = 8;
if (msgctl(qid,IPC_SET,&msg_attr) == -1)
{
      perror("msg set error!\n");
      exit(1);
}
getmsgattr(qid,msg_attr);/*修改后再观察其属性*/
if (msgctl(qid,IPC_RMID,NULL) == -1)
{
      perror("delete msg error!\n");
      exit(1);
}
getmsgattr(qid,msg_attr);/*删除后再观察其属性*/
}

void getmsgattr(int msgid,struct msqid_ds msg_info)
{
    if (msgctl(msgid,IPC_STAT,&msg_info) == -1)
    {
      perror("msgctl error!\n");
      return;
    }

    printf("***information of message queue %d***\n",msgid);
    printf("last msgsnd to msq time is %s\n",ctime(&(msg_info.msg_stime)));
    printf("last msgrcv time from msg is %s\n",ctime(&(msg_info.msg_rtime)));
    printf("last change msg time is %s\n",ctime(&(msg_info.msg_ctime)));
    printf("current number of bytes on queue is %d\n",msg_info.msg_cbytes);
    printf("number of messages in queue is %d\n",msg_info.msg_qnum);
    printf("max number of bytes on queue is %d\n",msg_info.msg_qbytes);
    printf("pid of last msgsnd is %d\n",msg_info.msg_lspid);
    printf("pid of last msgrcv is %d\n",msg_info.msg_lrpid);
    printf("msg uid is %d\n",msg_info.msg_perm.uid);
    printf("msg gid is %d\n",msg_info.msg_perm.gid);
    printf("*******information end!**************\n",msgid);
}

以上是对消息队列进行操作前的属性。发送消息后和重新设置后的消息队列属性都会因为操作而改变。可以运行程序观察全部的输出结果，对比操作前后消息队列属性是如何改变的。

四、消息队列的应用实例
这里以一个聊天程序为例，进一步展示消息队列的应用。
/*server*/

#include
#include
#include
#include
#include
#include
#include

#define BUF_SIZE   256
#define PROJ_ID    32
#define PATH_NAME  "/tmp"
#define SERVER_MSG 1
#define CLIENT_MSG 2

int main(void){
/*用户自定义消息缓冲区*/
struct mymsgbuf{
    long msgtype;
    char ctrlstring[BUF_SIZE];
} msgbuffer;
int qid;
int msglen;
key_t msgkey;

/*获取键值*/
if((msgkey=ftok(PATH_NAME,PROJ_ID))==-1)
{
      perror("ftok error!\n");
      exit(1);
}

/*获取消息队列标识符*/
if ((qid = msgget(msgkey,IPC_CREAT|0660)) == -1)
{
      perror("msgget error!\n");
      exit(1);
}

while(1)
{
  printf("server:");
  fgets(msgbuffer.ctrlstring,BUF_SIZE, stdin);
  if(strncmp("exit",msgbuffer.ctrlstring,4)==0)
  {
    msgctl(qid,IPC_RMID,NULL);
        break;

  }
  msgbuffer.ctrlstring[strlen(msgbuffer.ctrlstring) - 1] = '\0';
  msgbuffer.msgtype = SERVER_MSG;
  if (msgsnd(qid,&msgbuffer,strlen(msgbuffer.ctrlstring) + 1,0) == -1)
  {
    perror("Server msgsnd error!\n");
    exit(1);
  }

  if (msgrcv(qid,&msgbuffer,BUF_SIZE,CLIENT_MSG,0) == -1)
  {
    perror("Server msgrcv error!\n");
    exit(1);
  }
  printf("Client: %s\n",msgbuffer.ctrlstring);
}
exit(0);
}

#include 
#include 
#include 
#include 
#include 
#include 
#include 

#define BUF_SIZE   256
#define PROJ_ID    32
#define PATH_NAME  "/tmp"
#define SERVER_MSG 1
#define CLIENT_MSG 2

int main(void){
/*用户自定义消息缓冲区*/
struct mymsgbuf{
    long msgtype;
    char ctrlstring[BUF_SIZE];
} msgbuffer;
int qid;
int msglen;
key_t msgkey;

/*获取键值*/
if((msgkey=ftok(PATH_NAME,PROJ_ID))==-1)
{
      perror("ftok error!\n");
      exit(1);
}

/*获取消息队列标识符*/
if ((qid = msgget(msgkey,IPC_CREAT|0660)) == -1)
{
      perror("msgget error!\n");
      exit(1);
}

while(1)
{
  printf("server:");
  fgets(msgbuffer.ctrlstring,BUF_SIZE, stdin);
  if(strncmp("exit",msgbuffer.ctrlstring,4)==0)   
  {    
  	msgctl(qid,IPC_RMID,NULL);
        break;

  }
  msgbuffer.ctrlstring[strlen(msgbuffer.ctrlstring) - 1] = '\0';
  msgbuffer.msgtype = SERVER_MSG;
  if (msgsnd(qid,&msgbuffer,strlen(msgbuffer.ctrlstring) + 1,0) == -1)
  {
    perror("Server msgsnd error!\n");
    exit(1);
  }

  if (msgrcv(qid,&msgbuffer,BUF_SIZE,CLIENT_MSG,0) == -1)
  {
    perror("Server msgrcv error!\n");
    exit(1);
  }
  printf("Client: %s\n",msgbuffer.ctrlstring);
}
exit(0);
}

"client"

#include
#include
#include
#include
#include
#include
#include
#include

#define BUF_SIZE    256
#define PROJ_ID     32
#define PATH_NAME   "/tmp"
#define SERVER_MSG 1
#define CLIENT_MSG 2

int main(void)
{

/*用户自定义消息缓冲区*/
struct mymsgbuf{
    long msgtype;
    char ctrlstring[BUF_SIZE];
}msgbuffer;
int qid;/*消息队列标识符*/
int msglen;
key_t msgkey;

     /*获取键值*/
    if ((msgkey = ftok(PATH_NAME,PROJ_ID)) == -1)
    {
      perror("ftok error!\n");
      exit(1);
    }

    if ((qid = msgget(msgkey,IPC_CREAT|0660)) == -1)
    {
      perror("msgget error!\n");
      exit(1);
    }

    while(1)
    {
      if (msgrcv(qid,&msgbuffer,BUF_SIZE,SERVER_MSG,0) == -1)
      /*if queue is empty, block here*/
      {
           perror("Server msgrcv error!\n");
           exit(1);
      }

      printf("server: %s\n",msgbuffer.ctrlstring);
      printf("client:");
      fgets(msgbuffer.ctrlstring,BUF_SIZE,stdin);
      if (strncmp("exit",msgbuffer.ctrlstring,4) == 0)
      {
          break;
      }

      msgbuffer.ctrlstring[strlen(msgbuffer.ctrlstring)-1] = '\0';
      msgbuffer.msgtype = CLIENT_MSG;
      if (msgsnd(qid,&msgbuffer,strlen(msgbuffer.ctrlstring) + 1,0) == -1)
      {
          perror("client msgsnd error!\n");
          exit(1);
      }
    }
   exit(0);
}