Linux程序设计学习笔记----System V进程通信之消息队列

一个或多个进程可向消息队列写入消息,而一个或多个进程可从消息队列中读取消息,这种进程间通讯机制通常使用在客户/服务器模型中,客户向服务器发送请求消息,服务器读取消息并执行相应请求。在许多微内核结构的操作系统中,内核和各组件之间的基本通讯方式就是消息队列。例如,在 MINIX 操作系统中,内核、I/O 任务、服务器进程和用户进程之间就是通过消息队列实现通讯的。

Linux中的消息可以被描述成在内核地址空间的一个内部链表,每一个消息队列由一个IPC的标识号唯一的标识。Linux 为系统中所有的消息队列维护一个 msgque 链表,该链表中的每个指针指向一个 msgid_ds 结构,该结构完整描述一个消息队列。

System V消息队列是Open Group定义的XSI,不属于POSIX标准。System V IPC的历史相对很早,在上个世70年代后期有贝尔实验室的分支机构开发,80年代加入System V的系统内核中,后来商用UNIX系统基本都加入了System V IPC的功能。

System V消息队列相对于POSIX消息队列的区别主要是:

  • POSIX消息队列的读操作总是返回消息队列中优先级最高的最早消息,而对于System V消息队列可以返回任意指定优先级(通过消息类型)的消息。
  • 当向一个空消息队列中写入一个消息时,POSIX消息队列允许产生一个信号或启动一个线程,System V消息队列不提供类似的机制。

数据结构与基本属性

(1)消息缓冲区(msgbuf)

我们在这里要介绍的第一个数据结构是msgbuf结构,可以把这个特殊的数据结构看成一个存放消息数据的模板,它在include/linux/msg.h中声明,描述如下:

/* msgsnd 和msgrcv 系统调用使用的消息缓冲区*/

struct msgbuf {

   long mtype;        /* 消息的类型,必须为正数 */

   char mtext[1];      /* 消息正文 */

}; 

注意:对于消息数据元素(mtext),不要受其描述的限制。实际上,这个域(mtext)不仅能保存字符数组,而且能保存任何形式的任何数据。这个域本身是任意的,因为这个结构本身可以由应用程序员重新定义:

struct my_msgbuf {

        long    mtype;         /* 消息类型 */

       long   request_id;    /* 请求识别号 */

       struct  client info;    /* 客户消息结构 */

};
我们看到,消息的类型还是和前面一样,但是结构的剩余部分由两个其它的元素代替,而且有一个是结构。这就是消息队列的优美之处,内核根本不管传送的是什么样的数据,任何信息都可以传送。

但是,消息的长度还是有限制的,在Linux中,给定消息的最大长度在include/linux/msg.h中定义如下:

#define MSGMAX  8192    /* max size of message(bytes) */

消息总的长度不能超过8192字节,包括mtype域,它是4字节长。

(2)消息结构(msg)

内核把每一条消息存储在以msg结构为框架的队列中,它在include/ linux/msg.h中定义如下:

struct msg {

   struct msg *msg_next;  /* 队列上的下一条消息 */

   long  msg_type;         /*消息类型*/

   char *msg_spot;        /* 消息正文的地址 */

   short msg_ts;          /* 消息正文的大小 */

};
注意:msg_next是指向下一条消息的指针,它们在内核地址空间形成一个单链表。

(3)消息队列结构(msgid_ds)

当在系统中创建每一个消息队列时,内核创建、存储及维护这个结构的一个实例。

<bits/msq.h>

struct msqid_ds
{
  struct ipc_perm msg_perm;        /*IPC对象的属性信息和访问权限 */
  __time_t msg_stime;              /* time of last msgsnd command */
  __time_t msg_rtime;              /* time of last msgrcv command */
  __time_t msg_ctime;              /* time of last change */
  unsigned long int __msg_cbytes;  /* 当前队列中消息的字节数 */
  msgqnum_t msg_qnum;              /* 当前队列中消息的个数 */
  msglen_t msg_qbytes;             /* 队列允许存放的最大字节数 */
  __pid_t msg_lspid;               /* pid of last msgsnd() */
  __pid_t msg_lrpid;               /* pid of last msgrcv() */

//下面是保留字段
#if __WORDSIZE == 32
  unsigned long int __unused1;
  unsigned long int __unused2;
  unsigned long int __unused3;
#endif
  unsigned long int __unused4;
  unsigned long int __unused5;
};

System V消息队列的创建和打开

System V消息队列的创建和使用会使用下面的函数接口:

#include <sys/msg.h>
int msgget(key_t key, int oflg);
                //成功返回非负消息队列描述符,失败返回-1

key:消息队列的键,用来创建一个消息队列。System IPC都有一个key,作为IPC的外部标识符,创建成功后返回的描述符作为IPC的内部标识符使用。key的主要目的就是使不同进程在同一IPC汇合。key具体说可以有三种方式生成:

  • 不同的进程约定好的一个值;
  • 通过相同的路径名和项目ID,调用ftok()函数,生成一个键;
  • 还可以设置为IPC_PRIVATE,这样就会创建一个新的,唯一的IPC对象;然后将返回的描述符通过某种方式传递给其他进程;

oflg:指定创建或打开消息队列的标志和读写权限(ipc_perm中的mode成员)。我们知道System V IPC定义了自己的操作标志和权限设置标志,而且都是通过该参数传递,这和open函数存在差别,open函数第三个参数mode用于传递文件的权限标志。System V IPC的操作标志包含:IPC_CREATIPC_EXCL,权限设置标志如下图:


下面是创建消息队列的测试代码:

#include <iostream>
#include <cstring>
#include <errno.h>

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

using namespace std;

#define  PATH_NAME "/tmp/anonymQueue"

int main(int argc, char **argv)
{
    key_t key;
    int fd;

    if ((fd = open(PATH_NAME, O_CREAT, 0666)) < 0)
    {
        cout<<"open file "<<PATH_NAME<<"failed.";
        cout<<strerror(errno)<<endl;
        return -1;
    }
    close(fd);

    key = ftok(PATH_NAME, 0);
    int msgID;

    if ((msgID = msgget(key, IPC_CREAT | 0666)) < 0)
    {
        cout<<"open message queue failed...";
        cout<<strerror(errno)<<endl;
        return -1;
    }

    cout<<"key:0x"<<hex<<key<<endl;
    cout<<"descriptor id:"<<dec<<msgID<<endl;
}
结果:


实现System V IPC的任何系统都提供两个特殊的程序ipcsipcrmipcs输出IPC的各种信息,ipcrm则用于删除各种System V IPC。由于System V IPC不属于POSIX标准,所以这两个命令也未被标准化。

System V消息队列的使用

System V消息队列的写入消息使用下面的函数接口:

#include <sys/msg.h>
int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);
                        //成功返回0,失败返回-1

msqid:消息队列的描述符;

msgp:指向存放消息的缓冲区,该缓冲区中包含消息类型消息体两部分内容。该缓冲区的结构是由用户定义的,在<sys/msg.h>中有关于该缓冲区结构定义的参版考模

struct msgbuf
{
    long int mtype;             /* type of received/sent message */
    char mtext[1];              /* text of the message */
};

缓冲区的开头是一个long型的消息类型,该消息类型必须是一个非负数。紧跟在消息类型后面的是消息体部分(如果消息长度大于0),参考模版中定义的mtext只是说明消息体,该部分可以自定义长度。我们自己的应用都会定义特定的消息结构。

msgsz:缓冲区中消息体部分的长度;

msgflg:设置操作标志。可以为0IPC_NOWAIT用于在消息队列中没有可用的空间时,调用线程采用何种操作方式

标志为IPC_NOWAIT,表示msgsnd操作以非阻塞的方式进行,在消息队列中没有可用的空间时,msgsnd操作会立刻返回。并指定EAGAIN错误;

标志为0,表示msgsnd操作以阻塞的方式进行,这种情况下在消息队列中没有可用的空间时调用线程会被阻塞,直到下面的情况发生:

  • 等到有存放消息的空间;
  • 消息队列从系统中删除,这种情况下回返回一个EIDRM错误;
  • 调用线程被某个捕捉到的信号中断,这种情况下返回一个EINTR错误;

System V消息队列的读取消息使用下面的函数接口:

#include <sys/msg.h>
ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp, int msgflg);
                      //成功返回接收到的消息的消息体的字节数,失败返回-1

msqid:消息队列的描述符;

msgp:指向待存放消息的缓冲区,该缓冲区中将会存放接收到的消息的消息类型消息体两部分内容。该缓冲区的结构是由用户定义的,和msgsnd相对应。

msgsz:缓冲区中能存放消息体部分的最大长度,这也是该函数能返回的最大数据量;该字段的大小应该是sizeof(msg buffer) - sizeof(long)

msgtyp:希望从消息队列中获取的消息类型。

  • msgtyp0,返回消息队列中的第一个消息;
  • msgtyp > 0,返回该消息类型的第一个消息;
  • msgtyp < 0,返回小于或等于msgtyp绝对值的消息中类型最小的第一个消息;

msgflg:设置操作标志。可以为0IPC_NOWAITMSG_NOERROR用于在消息队列中没有可用的指定消息时,调用线程采用何种操作方式

标志为IPC_NOWAIT,表示msgrcv操作以非阻塞的方式进行,在消息队列中没有可用的指定消息时,msgrcv操作会立刻返回,并设定errnoENOMSG

标志为0,表示msgrcv操作是阻塞操作,直到下面的情况发生:

  • 消息队列中有一个所请求的消息类型可以获取;
  • 消息队列从系统中删除,这种情况下回返回一个EIDRM错误;
  • 调用线程被某个捕捉到的信号中断,这种情况下返回一个EINTR错误;

标志为MSG_NOERROR,表示接收到的消息的消息体的长度大于msgsz长度时,msgrcv采取的操作。如果设置了该标志msgrcv在这种情况下回截断数据部分,而不返回错误,否则返回一个E2BIG错误。

下面是关于消息队列读写的测试代码:

#include <iostream>
#include <cstring>
#include <errno.h>

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

using namespace std;

#define  PATH_NAME "/tmp/anonymQueue"

key_t CreateKey(const char *pathName)
{
    int fd;

    if ((fd = open(PATH_NAME, O_CREAT, 0666)) < 0)
    {
        cout<<"open file "<<PATH_NAME<<"failed.";
        cout<<strerror(errno)<<endl;
        return -1;
    }

    close(fd);

    return ftok(PATH_NAME, 0);
}

int main(int argc, char **argv)
{
    key_t key;
    key = CreateKey(PATH_NAME);

    int msgID;
    if ((msgID = msgget(key, IPC_CREAT | 0666)) < 0)
    {
        cout<<"open message queue failed...";
        cout<<strerror(errno)<<endl;

        return -1;
    }

    if (fork() == 0)
    {
        char msg[] = "cute yuki...";
        char *msgBuf = new char[sizeof(long) + sizeof(msg)];
        long mtype = 1;

        memmove(msgBuf, &mtype, sizeof(mtype));
        memmove(msgBuf + sizeof(mtype), msg, sizeof(msg));

        for (int i = 1; i <= 5; ++i)
        {
            if (msgsnd(msgID, msgBuf, sizeof(msg), 0) < 0)
            {
                cout<<"send message "<<i<<"failed...";
                cout<<strerror(errno)<<endl;
                continue;
            }

            cout<<"child: send message "<<i<<" success..."<<endl;
            sleep(1);
        }

        exit(0);
    }

    char msgBuf[256];
    long mtype;
    int recvLen;

    for (int i = 1; i <= 5; ++i)
    {
        recvLen = msgrcv(msgID, msgBuf, 256 - sizeof(long), 0, 0);
        if (recvLen < 0)
        {
            cout<<"receive message failed...";
            cout<<strerror(errno)<<endl;
            continue;
        }

        memmove(&mtype, msgBuf, sizeof(long));

        cout<<"parent receive a message:"<<endl;
        cout<<"message type:"<<mtype<<endl;
        cout<<"message body:"<<msgBuf + sizeof(long)<<endl;
    }
    return 0;
}

System V消息队列的控制操作

System V消息队列的删除,属性的设置和获取等控制操作要使用下面的函数接口:

  1. #include <sys/msg.h>  
  2. int msgctl(int msqid, int cmd, struct msqid_ds *buf);  
  3.                         //成功返回0,失败返回-1  

msqid:消息队列的描述符;

cmd:控制操作的命令,SUS标准提供以下三个命令:

  • IPC_RMID,删除一个消息队列。执行该命令系统会立刻把该消息队列从内核中删除,该消息队列中的所有消息将会被丢弃。这和已经讨论过的POSIX消息队列有很大差别POSIX消息队列通过调用mq_unlink来从内核中删除一个消息队列,但消息队列的真正析构会在最后一个mq_close结束后发生。
  •  IPC_SET,根据buf的所指的值来设置消息队列msqid_ds结构中的msg_perm.uidmsg_perm.gidmsg_perm.modemsg_qbytes四个成员。
  • IPC_STAT,通过buf返回当前消息队列的msqid_ds结构。
  • Linux下还有例如IPC_INFOMSG_INFO等命令,具体可以参考Linux手册;

buf:指向msqid_ds结构的指针;

下面是测试代码:

  1. #include <iostream>  
  2. #include <cstring>  
  3. #include <errno.h>  
  4.   
  5. #include <unistd.h>  
  6. #include <fcntl.h>  
  7. #include <sys/msg.h>  
  8.   
  9. using namespace std;  
  10.   
  11. #define  PATH_NAME "/tmp/anonymQueue"  
  12.   
  13. key_t CreateKey(const char *pathName)  
  14. {  
  15.     int fd;  
  16.     if ((fd = open(PATH_NAME, O_CREAT, 0666)) < 0)  
  17.     {  
  18.         cout<<"open file "<<PATH_NAME<<"failed.";  
  19.         cout<<strerror(errno)<<endl;  
  20.         return -1;  
  21.     }  
  22.   
  23.     close(fd);  
  24.   
  25.     return ftok(PATH_NAME, 0);  
  26. }  
  27.   
  28. int main(int argc, char **argv)  
  29. {  
  30.     key_t key;  
  31.     key = CreateKey(PATH_NAME);  
  32.   
  33.     int msgID;  
  34.     if ((msgID = msgget(key, IPC_CREAT | 0666)) < 0)  
  35.     {  
  36.         cout<<"open message queue failed...";  
  37.         cout<<strerror(errno)<<endl;  
  38.         return -1;  
  39.     }  
  40.   
  41.     msqid_ds msgInfo;  
  42.     msgctl(msgID, IPC_STAT, &msgInfo);  
  43.   
  44.     cout<<"msg_qbytes:"<<msgInfo.msg_qbytes<<endl;  
  45.     cout<<"msg_qnum:"<<msgInfo.msg_qnum<<endl;  
  46.     cout<<"msg_cbytes:"<<msgInfo.msg_cbytes<<endl;  
  47.     return 0;  
  48. }  

Linux 2.6.18下的执行结果为:

  1. msg_qbytes:65536  
  2. msg_qnum:2  
  3. msg_cbytes:26  

关于消息队列中允许存放最大的字节数可以通过IPC_SET命令进行修改,该修改只能针对本消息队列生效。如下测试代码:

  1. int main(int argc, char **argv)  
  2. {  
  3.     key_t key;  
  4.     key = CreateKey(PATH_NAME);  
  5.   
  6.     int msgID;  
  7.     if ((msgID = msgget(key, IPC_CREAT | 0666)) < 0)  
  8.     {  
  9.         cout<<"open message queue failed...";  
  10.         cout<<strerror(errno)<<endl;  
  11.         return -1;  
  12.     }  
  13.   
  14.     msqid_ds msgInfo;  
  15.     msgctl(msgID, IPC_STAT, &msgInfo);  
  16.   
  17.     cout<<"msg_qbytes:"<<msgInfo.msg_qbytes<<endl;  
  18.     cout<<"msg_qnum:"<<msgInfo.msg_qnum<<endl;  
  19.     cout<<"msg_cbytes:"<<msgInfo.msg_cbytes<<endl;  
  20.   
  21.     msgInfo.msg_qbytes = 6553600;  
  22.     if (msgctl(msgID, IPC_SET, &msgInfo) < 0)  
  23.     {  
  24.         cout<<"set message queue failed...";  
  25.         cout<<strerror(errno)<<endl;  
  26.         return -1;  
  27.     }  
  28.   
  29.     msgctl(msgID, IPC_STAT, &msgInfo);  
  30.     cout<<"msg_qbytes:"<<msgInfo.msg_qbytes<<endl;  
  31.     cout<<"msg_qnum:"<<msgInfo.msg_qnum<<endl;  
  32.     cout<<"msg_cbytes:"<<msgInfo.msg_cbytes<<endl;  
  33.     return 0;  
  34. }  

Linux 2.6.18下的执行结果为:

  1. msg_qbytes:65536  
  2. msg_qnum:0  
  3. msg_cbytes:0  
  4. msg_qbytes:6553600  
  5. msg_qnum:0  
  6. msg_cbytes:0  

 System V消息队列的内核限制

System V IPC,系统往往会存在一些限制,对于消息队列,在Linux2.6.18中,系统内核存在以下限制:

  1. [root@idcserver program]# sysctl -a |grep msg  
  2. kernel.msgmnb = 65536 //一个消息队列上允许的最大字节数  
  3. kernel.msgmni = 16 //系统范围内允许存在的最大消息队列数  
  4. kernel.msgmax = 65536     //每个消息的最大字节数  

对于System V消息队列一般内核还有一个限制:系统范围内的最大消息数,在Linux下这个限制由msgmnb*msgmni决定。

上面已经说过可以通过IPC_SET来设置使用中的消息队列的最大字节数。但是要在系统范围内对内核限制进行修改,在Linux下面可以通过修改/etc/sysctl.conf内核参数配置文件,然后配合sysctl命令来对内核参数进行设置。例如下面示例:

  1. [root@idcserver program]#echo "kernel.msgmnb = 6553600" >>/etc/sysctl.conf  
  2. [root@idcserver program]#echo "kernel.msgmni = 100" >>/etc/sysctl.conf  
  3. [root@idcserver program]#echo "kernel.msgmax = 6553600" >>/etc/sysctl.conf  
  4. [root@idcserver program]#sysctl -p  
  5. [root@idcserver program]# sysctl -a |grep msg  
  6. kernel.msgmnb = 6553600  
  7. kernel.msgmni = 100  
  8. kernel.msgmax = 6553600 





Linux程序设计学习笔记----System V进程通信之消息队列,古老的榕树,5-wow.com

郑重声明:本站内容如果来自互联网及其他传播媒体,其版权均属原媒体及文章作者所有。转载目的在于传递更多信息及用于网络分享,并不代表本站赞同其观点和对其真实性负责,也不构成任何其他建议。