Linux进程通信之POSIX消息队列

绝地灬酷狼 2022-07-14 04:00 200阅读 0赞

消息队列是 Linux IPC 中很常用的一种通信方式，它通常用来在不同进程间发送特定格式的消息数据。

消息队列和之前讨论过的 [管道和 FIFO ][FIFO]有很大的区别，主要有以下两点：

*  一个进程向消息队列写入消息之前，并不需要某个进程在该队列上等待该消息的到达，而管道和 FIFO 是相反的，进程向其中写消息时，管道和 FIFO 必需已经打开来读，否则写进程就会阻塞（默认情况下）。
 *  IPC 的持续性不同。管道和 FIFO 是 **随进程**的持续性，当管道和 FIFO 最后一次关闭发生时，仍在管道和 FIFO 中的数据会被丢弃。消息队列是 **随内核**的持续性，即一个进程向消息队列写入消息后，然后终止，另外一个进程可以在以后某个时刻打开该队列读取消息。只要内核没有重新自举，消息队列没有被删除。

消息队列中的每条消息通常具有以下属性：

*  一个表示优先级的整数；
 *  消息的数据部分的长度；
 *  消息数据本身；

POSIX 消息队列的一个可能的设计是一个如下图所示的消息链表，链表头部有消息队列的属性信息。

![Zr6bqq7.jpg_web][]

图 1 消息队列的可能布局

### 1 POSIX 消息队列的创建和关闭 ###

POSIX 消息队列的创建，关闭和删除用到以下三个函数接口：

#include <mqueue.h>
    mqd_t mq_open(const char *name, int oflag, /* mode_t mode, struct mq_attr *attr */);
                           //成功返回消息队列描述符，失败返回-1
    mqd_t mq_close(mqd_t mqdes);
    mqd_t mq_unlink(const char *name);
                               //成功返回0，失败返回-1

mq\_open 用于打开或创建一个消息队列。

name ：表示消息队列的名字，它符合 POSIX IPC 的名字规则。

oflag ：表示打开的方式，和 open 函数的类似。有必须的选项： O\_RDONLY ， O\_WRONLY ， O\_RDWR ，还有可选的选项： O\_NONBLOCK ， O\_CREAT ， O\_EXCL。

mode ：是一个可选参数，在 oflag 中含有 O\_CREAT 标志且消息队列不存在时，才需要提供该参数。表示默认访问权限。可以参考 open 。

attr ：也是一个可选参数，在 oflag 中含有 O\_CREAT 标志且消息队列不存在时才需要。该参数用于给新队列设定某些属性，如果是空指针，那么就采用默认属性。

mq\_open 返回值是 mqd\_t 类型的值，被称为消息队列描述符。在 Linux  2.6.18中该类型的定义为整型：

#include <bits/mqueue.h>
    typedef int mqd_t;

mq\_close 用于关闭一个消息队列，和文件的 close 类型，关闭后，消息队列并不从系统中删除。一个进程结束，会自动调用关闭打开着的消息队列。

mq\_unlink 用于删除一个消息队列。消息队列创建后只有通过调用该函数或者是内核自举才能进行删除。每个消息队列都有一个保存当前打开着描述符数的引用计数器，和文件一样，因此本函数能够实现类似于 unlink 函数删除一个文件的机制。

POSIX 消息队列的名字 所创建的真正路径名和具体的系统实现有关，关于具体 POSIX IPC 的名字规则可以参考《 UNIX  网络编程 卷 2 ：进程间通信》的 P14 。

经过测试，在 Linux  2.6.18中，所创建的POSIX 消息队列不会在文件系统中创建真正的路径名。且 POSIX 的名字只能以一个 ’/’开头，名字中不能包含其他的’/’。

### 2 POSIX 消息队列的属性 ###

POSIX 标准规定消息队列属性 mq\_attr 必须要含有以下四个内容：

long    mq_flags //消息队列的标志：0或O_NONBLOCK,用来表示是否阻塞 
    long    mq_maxmsg  //消息队列的最大消息数
    long    mq_msgsize  //消息队列中每个消息的最大字节数
    long    mq_curmsgs  //消息队列中当前的消息数目

在 Linux  2.6.18中mq\_attr 结构的定义如下：

#include <bits/mqueue.h>
    struct mq_attr
    {
      long int mq_flags;      /* Message queue flags.  */
      long int mq_maxmsg;   /* Maximum number of messages.  */
      long int mq_msgsize;   /* Maximum message size.  */
      long int mq_curmsgs;   /* Number of messages currently queued.  */
      long int __pad[4];
    };

POSIX 消息队列的属性设置和获取可以通过下面两个函数实现：

#include <mqueue.h>
    mqd_t mq_getattr(mqd_t mqdes, struct mq_attr *attr);
    mqd_t mq_setattr(mqd_t mqdes, struct mq_attr *newattr, struct mq_attr *oldattr);
                                   //成功返回0，失败返回-1

mq\_getattr 用于获取当前消息队列的属性， mq\_setattr 用于设置当前消息队列的属性。其中 mq\_setattr 中的 oldattr 用于保存修改前的消息队列的属性，可以为空。

mq\_setattr 可以设置的属性只有 mq\_flags，用来设置或清除消息队列的非阻塞标志。 newattr 结构的其他属性被忽略。 mq\_maxmsg 和 mq\_msgsize 属性只能在创建消息队列时通过 mq\_open 来设置。 mq\_open 只会设置该两个属性，忽略另外两个属性。 mq\_curmsgs 属性只能被获取而不能被设置。

下面是测试代码：

#include <iostream>
    #include <cstring>
    
    #include <errno.h>
    #include <unistd.h>
    #include <fcntl.h>
    #include <mqueue.h>
    
    using namespace std;
    
    int main()
    {
        mqd_t mqID;
        mqID = mq_open("/anonymQueue", O_RDWR | O_CREAT, 0666, NULL);
    
        if (mqID < 0)
        {
            cout<<"open message queue error..."<<strerror(errno)<<endl;
            return -1;
        }
    
        mq_attr mqAttr;
        if (mq_getattr(mqID, &mqAttr) < 0)
        {
            cout<<"get the message queue attribute error"<<endl;
            return -1;
        }
    
        cout<<"mq_flags:"<<mqAttr.mq_flags<<endl;
        cout<<"mq_maxmsg:"<<mqAttr.mq_maxmsg<<endl;
        cout<<"mq_msgsize:"<<mqAttr.mq_msgsize<<endl;
        cout<<"mq_curmsgs:"<<mqAttr.mq_curmsgs<<endl;
    }

在 Linux  2.6.18中执行结果是：

mq_flags:0
    mq_maxmsg:10
    mq_msgsize:8192
    mq_curmsgs:0

### 3 POSIX 消息队列的使用 ###

POSIX 消息队列可以通过以下两个函数来进行发送和接收消息：

#include <mqueue.h>
    mqd_t mq_send(mqd_t mqdes, const char *msg_ptr,
                          size_t msg_len, unsigned msg_prio);
                         //成功返回0，出错返回-1
    
    mqd_t mq_receive(mqd_t mqdes, char *msg_ptr,
                          size_t msg_len, unsigned *msg_prio);
                         //成功返回接收到消息的字节数，出错返回-1
    
    #ifdef __USE_XOPEN2K
    mqd_t mq_timedsend(mqd_t mqdes, const char *msg_ptr,
                          size_t msg_len, unsigned msg_prio,
                          const struct timespec *abs_timeout);
    
    mqd_t mq_timedreceive(mqd_t mqdes, char *msg_ptr,
                          size_t msg_len, unsigned *msg_prio,
                          const struct timespec *abs_timeout);
    #endif

mq\_send 向消息队列中写入一条消息， mq\_receive 从消息队列中读取一条消息。

mqdes ：消息队列描述符；

msg\_ptr ：指向消息体缓冲区的指针；

msg\_len ：消息体的长度，其中 mq\_receive 的该参数不能小于能写入队列中消息的最大大小，即一定要大于等于该队列的 mq\_attr 结构中 mq\_msgsize 的大小。如果 mq\_receive 中的 msg\_len 小于该值，就会返回 EMSGSIZE 错误。POXIS 消息队列发送的消息长度可以为 0 。

msg\_prio ：消息的优先级；它是一个小于 MQ\_PRIO\_MAX 的数，数值越大，优先级越高。 POSIX 消息队列在调用 mq\_receive 时总是返回队列中 最高优先级的最早消息。如果消息不需要设定优先级，那么可以在 mq\_send 是置 msg\_prio 为 0 ， mq\_receive的 msg\_prio 置为 NULL 。

还有两个XSI定义的扩展接口限时发送和接收消息的函数： mq\_timedsend 和 mq\_timedreceive 函数。默认情况下 mq\_send 和 mq\_receive 是阻塞进行调用，可以通过 mq\_setattr来设置为 O\_NONBLOCK 。

下面是消息队列使用的测试代码：

#include <iostream>
    #include <cstring>
    #include <errno.h>
    
    #include <unistd.h>
    #include <fcntl.h>
    #include <mqueue.h>
    
    using namespace std;
    
    int main()
    {
        mqd_t mqID;
        mqID = mq_open("/anonymQueue", O_RDWR | O_CREAT | O_EXCL, 0666, NULL);
    
        if (mqID < 0)
        {
            if (errno == EEXIST)
            {
                mq_unlink("/anonymQueue");
                mqID = mq_open("/anonymQueue", O_RDWR | O_CREAT, 0666, NULL);
            }
            else
            {
                cout<<"open message queue error..."<<strerror(errno)<<endl;
                return -1;
            }
        }
    
        if (fork() == 0)
        {
            mq_attr mqAttr;
            mq_getattr(mqID, &mqAttr);
    
            char *buf = new char[mqAttr.mq_msgsize];
    
            for (int i = 1; i <= 5; ++i)
            {
                if (mq_receive(mqID, buf, mqAttr.mq_msgsize, NULL) < 0)
                {
                    cout<<"receive message  failed. ";
                    cout<<"error info:"<<strerror(errno)<<endl;
                    continue;
                }
    
                cout<<"receive message "<<i<<": "<<buf<<endl;   
            }
            exit(0);
        }
    
        char msg[] = "yuki";
        for (int i = 1; i <= 5; ++i)
        {
            if (mq_send(mqID, msg, sizeof(msg), i) < 0)
            {
                cout<<"send message "<<i<<" failed. ";
                cout<<"error info:"<<strerror(errno)<<endl;
            }
            cout<<"send message "<<i<<" success. "<<endl;   
    
            sleep(1);
        }
    }

在 Linux  2.6.18下的执行结构如下：

send message 1 success. 
    receive message 1: yuki
    send message 2 success. 
    receive message 2: yuki
    send message 3 success. 
    receive message 3: yuki
    send message 4 success. 
    receive message 4: yuki
    send message 5 success. 
    receive message 5: yuki

### 4 POSIX 消息队列的限制 ###

POSIX 消息队列本身的限制就是 mq\_attr 中的 mq\_maxmsg 和 mq\_msgsize ，分别用于限定消息队列中的最大消息数和每个消息的最大字节数。在前面已经说过了，这两个参数可以在调用 mq\_open 创建一个消息队列的时候设定。当这个设定是受到系统内核限制的。

下面是在 Linux  2.6.18下 shell 对启动进程的 POSIX 消息队列大小的限制：

# ulimit -a |grep message
    POSIX message queues     (bytes, -q) 819200

限制大小为 800KB ， 该大小是整个消息队列的大小，不仅仅是最大消息数 \* 消息的最大大小；还包括消息队列的额外开销。前面我们知道 Linux  2.6.18下POSIX 消息队列默认的最大消息数和消息的最大大小分别为：

mq_maxmsg = 10
    mq_msgsize = 8192

为了说明上面的限制大小包括消息队列的额外开销，下面是测试代码：

#include <iostream>
    #include <cstring>
    #include <errno.h>
    
    #include <unistd.h>
    #include <fcntl.h>
    #include <mqueue.h>
    
    using namespace std;
    
    int main(int argc, char **argv)
    {
        mqd_t mqID;
        mq_attr attr;
        attr.mq_maxmsg = atoi(argv[1]);
        attr.mq_msgsize = atoi(argv[2]);
    
        mqID = mq_open("/anonymQueue", O_RDWR | O_CREAT | O_EXCL, 0666, &attr);
    
        if (mqID < 0)
        {
            if (errno == EEXIST)
            {
                mq_unlink("/anonymQueue");
                mqID = mq_open("/anonymQueue", O_RDWR | O_CREAT, 0666, &attr);
    
                if(mqID < 0)
                {
                    cout<<"open message queue error..."<<strerror(errno)<<endl;
                    return -1;
                }
            }
            else
            {
                cout<<"open message queue error..."<<strerror(errno)<<endl;
                return -1;
            }
        }
    
        mq_attr mqAttr;
        if (mq_getattr(mqID, &mqAttr) < 0)
        {
            cout<<"get the message queue attribute error"<<endl;
            return -1;
        }
    
        cout<<"mq_flags:"<<mqAttr.mq_flags<<endl;
        cout<<"mq_maxmsg:"<<mqAttr.mq_maxmsg<<endl;
        cout<<"mq_msgsize:"<<mqAttr.mq_msgsize<<endl;
        cout<<"mq_curmsgs:"<<mqAttr.mq_curmsgs<<endl; 
    }

下面进行创建消息队列时设置最大消息数和消息的最大大小进行测试：

[root@idcserver program]# g++ -g test.cpp -lrt
    [root@idcserver program]# ./a.out 10 81920
    open message queue error...Cannot allocate memory
    [root@idcserver program]# ./a.out 10 80000
    open message queue error...Cannot allocate memory
    [root@idcserver program]# ./a.out 10 70000
    open message queue error...Cannot allocate memory
    [root@idcserver program]# ./a.out 10 60000
    mq_flags:0
    mq_maxmsg:10
    mq_msgsize:60000
    mq_curmsgs:0

从上面可以看出消息队列真正存放消息数据的大小是没有 819200B 的。可以通过修改该限制参数，来改变消息队列的所能容纳消息的数量。可以通过下面方式来修改限制，但这会在 shell 启动进程结束后失效，可以将设置写入开机启动的脚本中执行，例如 .bashrc ， rc.local 。

[root@idcserver ~]# ulimit -q 1024000000
    [root@idcserver ~]# ulimit -a |grep message
    POSIX message queues     (bytes, -q) 1024000000

下面再次测试可以设置的消息队列的属性。

[root@idcserver program]# ./a.out 10 81920
    mq_flags:0
    mq_maxmsg:10
    mq_msgsize:81920
    mq_curmsgs:0
    [root@idcserver program]# ./a.out 10 819200
    mq_flags:0
    mq_maxmsg:10
    mq_msgsize:819200
    mq_curmsgs:0
    [root@idcserver program]# ./a.out 1000 8192  
    mq_flags:0
    mq_maxmsg:1000
    mq_msgsize:8192
    mq_curmsgs:0

POSIX 消息队列在实现上还有另外两个限制：

MQ\_OPEN\_MAX ：一个进程能同时打开的消息队列的最大数目，POSIX要求至少为8；

MQ\_PRIO\_MAX ：消息的最大优先级，POSIX要求至少为32；

Aug 7, 2013 AM 08:53 @lab

[FIFO]: http://blog.csdn.net/anonymalias/article/details/9391743
[Zr6bqq7.jpg_web]: http://img1.tuicool.com/Zr6bqq7.jpg!web