IPC之Posix消息队列详解

客官°小女子只卖身不卖艺 2022-07-17 02:39 255阅读 0赞

**基本概念：**

消息队列可认为是一个消息链表。有足够写权限的线程可往队列中放置消息，有足够读权限的线程可从队列中取走消息，每个消息都是一个记录（非字节流式，也就是不需要自定义边界），它由发送者赋予一个优先级。在某个进程往一个队列写入消息之前，并不需要另外某个进程在该队列上等待消息的到达。

一个进程可以往某个队列写入一些消息，然后终止，再让另外一个进程在以后某个时刻读出这些消息。消息队列具有随内核的持续性，也就是进程关闭后，消息队列依然存在。

**消息队列的标准：**

消息队列分为两大标准，Posix消息队列和System V消息队列，此文只讲Posix消息队列。

**此文的环境：**

例子的运行系统：Ubuntu 14

运行前需要执行的命令（为什么需要执行下列命令，请查阅**man 7 mq\_overview**）：

Mounting the message queue filesystem
           On Linux, message queues are created in a virtual  filesystem.   (Other
           implementations  may  also  provide such a feature, but the details are
           likely to differ.)  This filesystem can be mounted (by  the  superuser)
           using the following commands:
    
    
               # mkdir /dev/mqueue
               # mount -t mqueue none /dev/mqueue
    
    
           The sticky bit is automatically enabled on the mount directory.

**创建消息队列：**

MQ_OPEN(3)                 Linux Programmer's Manual                MQ_OPEN(3)
    
    NAME
           mq_open - open a message queue
    
    SYNOPSIS
           #include <fcntl.h>           /* For O_* constants */
           #include <sys/stat.h>        /* For mode constants */
           #include <mqueue.h>
    
           mqd_t mq_open(const char *name, int oflag);
           mqd_t mq_open(const char *name, int oflag, mode_t mode,
                         struct mq_attr *attr);
    
           Link with -lrt.

返回：若成功则为消息队列描述符，若出错则为-1

**关闭消息队列：**

MQ_CLOSE(3)                Linux Programmer's Manual               MQ_CLOSE(3)
    
    NAME
           mq_close - close a message queue descriptor
    
    SYNOPSIS
           #include <mqueue.h>
    
           int mq_close(mqd_t mqdes);
    
           Link with -lrt.

返回：若成功则为0，若出错则为-1

**删除消息队列：**

MQ_UNLINK(3)               Linux Programmer's Manual              MQ_UNLINK(3)
    
    NAME
           mq_unlink - remove a message queue
    
    SYNOPSIS
           #include <mqueue.h>
    
           int mq_unlink(const char *name);
    
           Link with -lrt.

返回：若成功则为0，若出错则为-1

**获取/设置消息队列属性：**

MQ_GETATTR(3)              Linux Programmer's Manual             MQ_GETATTR(3)
    
    NAME
           mq_getattr, mq_setattr - get/set message queue attributes
    
    SYNOPSIS
           #include <mqueue.h>
    
           int mq_getattr(mqd_t mqdes, struct mq_attr *attr);
    
           int mq_setattr(mqd_t mqdes, struct mq_attr *newattr,
                            struct mq_attr *oldattr);
    
           Link with -lrt.

均返回：若成功则为0，若出错则为-1

mq\_attr结构含有以下属性

struct mq_attr {
    	long mq_flags;       /* Flags: 0 or O_NONBLOCK */
    	long mq_maxmsg;      /* Max. # of messages on queue */
    	long mq_msgsize;     /* Max. message size (bytes) */
    	long mq_curmsgs;     /* # of messages currently in queue */
    };

a)mq\_setattr函数只允许设置mq\_attr结构的mq\_flags成员，其它三个成员被忽略。

b)指向某个mq\_attr结构的指针可作为mq\_open的第四个参数传递，每个队列的最大消息数和每个消息的最大字节数只能在创建队列时设置，而且这两者必须同时指定。

c)消息队列中的当前消息数则只能获取不能设置。

**例子1，创建并设置消息队列的属性，再删除消息队列。**

#include <fcntl.h>
    #include <sys/stat.h>
    #include <mqueue.h>
    #include <stdio.h>
    #include <errno.h>
    #include <stdlib.h>
    #include <string.h>
    
    int main()
    {
        struct mq_attr attr;
        attr.mq_maxmsg = 24; /* 设置消息队列的最大消息个数 */
        attr.mq_msgsize = 8192; /* 设置每个消息的最大字节数 */
    
        char *name = "/temp.mq"; /* linux必须是/filename这种格式，不能出现二级目录 */
        mqd_t mqd = mq_open(name, O_RDWR | O_CREAT | O_EXCL, 0777, &attr);
        if (mqd == -1) {
            perror("create failed");
            mqd = mq_open(name, O_RDWR);
            if (mqd == -1) {
                perror("open failed");
                exit(EXIT_FAILURE);
            }
        }
        printf("mq_open %s success\n", name);
    
        /* 打开成功，获取当前属性 */
        mq_getattr(mqd, &attr);
        printf("max msg = %ld, max bytes = %ld, currently = %ld\n",
                attr.mq_maxmsg, attr.mq_msgsize, attr.mq_curmsgs);
    
        /* 关闭 */
        if (mq_close(mqd) != -1)
        	printf("mq_close %s success\n", name);
    
        /* 删除 */
        sleep(30); /* 观察创建情况 */
        if (mq_unlink(name) != -1)
            printf("mq_unlink %s success\n", name);
    
        return 0;
    }

**编译运行：**

![20160827184413911][]

**发送消息：**

MQ_SEND(3)                 Linux Programmer's Manual                MQ_SEND(3)
    
    NAME
           mq_send, mq_timedsend - send a message to a message queue
    
    SYNOPSIS
           #include <mqueue.h>
    
           int mq_send(mqd_t mqdes, const char *msg_ptr,
                         size_t msg_len, unsigned msg_prio);
    
           #include <time.h>
           #include <mqueue.h>
    
           int mq_timedsend(mqd_t mqdes, const char *msg_ptr,
                         size_t msg_len, unsigned msg_prio,
                         const struct timespec *abs_timeout);
    
           Link with -lrt.

返回：若成功则为0，若出错则为-1

mq\_send的prio参数是待发送消息的优先级，其值必须小于MQ\_PRIO\_MAX。如果应用不必使用优先级不同的消息，那就给mq\_send指定值为0的优先级，给mq\_reveive指定一个空指针作为其最后一个参数。

**接收消息：**

MQ_RECEIVE(3)              Linux Programmer's Manual             MQ_RECEIVE(3)
    
    NAME
           mq_receive, mq_timedreceive - receive a message from a message queue
    
    SYNOPSIS
           #include <mqueue.h>
    
           ssize_t mq_receive(mqd_t mqdes, char *msg_ptr,
                              size_t msg_len, unsigned *msg_prio);
    
           #include <time.h>
           #include <mqueue.h>
    
           ssize_t mq_timedreceive(mqd_t mqdes, char *msg_ptr,
                              size_t msg_len, unsigned *msg_prio,
                              const struct timespec *abs_timeout);
    
           Link with -lrt.

返回：若成功则为消息中接字数，若出错则为-1

mq\_receive总是返回所指定队列中最高优先级的最早消息，而且该优先级能随该消息的内容及其长度一同返回。

mq\_receive的len参数的值不能小于能加到所指定队里中的消息的最大大小（该队列mq\_attr结构的mq\_msgsize成员）。要是len小于该值，mq\_reveive就立即返回EMSGSIZE错误。这意味着使用Posix消息队列的大多数应用程序必须在打开某个队列后调用mq\_getattr确定最大消息大小，然后分配一个或多个那样大小的读缓冲区。

**例子2，发送3个优先级不同的消息并接收。**

**发送程序mqsend.c：**

#include <fcntl.h>
    #include <sys/stat.h>
    #include <mqueue.h>
    #include <stdio.h>
    #include <errno.h>
    #include <stdlib.h>
    #include <string.h>
    
    int main()
    {
        struct mq_attr attr;
        attr.mq_maxmsg = 24; /* 设置消息队列的最大消息个数 */
        attr.mq_msgsize = 8192; /* 设置每个消息的最大字节数 */
    
        char *name = "/temp.mq"; /* linux必须是/filename这种格式，不能出现二级目录 */
        mqd_t mqd = mq_open(name, O_RDWR | O_CREAT | O_EXCL, 0777, &attr);
        if (mqd == -1) {
            perror("create failed");
            mqd = mq_open(name, O_RDWR);
            if (mqd == -1) {
                perror("open failed");
                exit(EXIT_FAILURE);
            }
        }
        printf("mq_open %s success\n", name);
    
        /* 打开成功，获取当前属性 */
        mq_getattr(mqd, &attr);
        printf("max msg = %ld, max bytes = %ld, currently = %ld\n",
                attr.mq_maxmsg, attr.mq_msgsize, attr.mq_curmsgs);
    
        char *msg_ptr1 = "hello world1";
        char *msg_ptr2 = "hello world2";
        char *msg_ptr3 = "hello world3";
        size_t msg_len1 = strlen(msg_ptr1);
        size_t msg_len2 = strlen(msg_ptr2);
        size_t msg_len3 = strlen(msg_ptr3);
    
        /* 先发送优先级低的 */
        mq_send(mqd, msg_ptr1, msg_len1, 1);
        mq_send(mqd, msg_ptr2, msg_len2, 2);
        mq_send(mqd, msg_ptr3, msg_len3, 3);
    
        printf("mq_send success\n");
        if (mq_close(mqd) != -1)
            printf("mq_close %s success\n", name);
    
        return 0;
    }

**接收程序mqreceive.c：**

#include <fcntl.h>
    #include <sys/stat.h>
    #include <mqueue.h>
    #include <stdio.h>
    #include <errno.h>
    #include <stdlib.h>
    #include <string.h>
    
    int main()
    {
        char *name = "/temp.mq"; /* linux必须是/filename这种格式，不能出现二级目录 */
        mqd_t mqd = mq_open(name, O_RDWR);
        if (mqd == -1) {
            perror("open failed");
            exit(EXIT_FAILURE);
        }
        printf("mq_open %s success\n", name);
    
        struct mq_attr attr;
        if (mq_getattr(mqd, &attr) == -1) {
            perror("mq_getattr failed");
            exit(EXIT_FAILURE);
        }
    
        char *msg_ptr;
        size_t msg_len = attr.mq_msgsize;
        unsigned msg_prio;
        msg_ptr = (char *)malloc(msg_len);
    
        while (1) {
            bzero(msg_ptr, msg_len);
            int res = mq_receive(mqd, msg_ptr, msg_len, &msg_prio);
            if (res != -1) {
                printf("msg is:%s, msg_prio:%d\n", msg_ptr, msg_prio);
            } else {
                perror("mq_receive failed");
                break;
            }
        }
    
        if (mq_close(mqd) != -1)
            printf("mq_close %s success\n", name);
    
        return 0;
    }

**编译运行：**

![20160827203146610][]

消息队列限制：

我们已经遇到任意给定队列的两个限制，它们都是在创建该队列时建立的（测试中发现mq\_getattr与ulimit -q的结果并不一致）：

mq\_maxmsg ->队列中的最大消息数

mq\_msgsize ->给定消息的最大字节数

消息队列的实现定义了另外两个限制（调用sysconf函数获取）：

MQ\_OPEN\_MAX -> 一个进程能够同时拥有的打开着消息队列的最大数目

MQ\_PRIO\_MAX -> 任意消息的最大优先级加1

**测试代码与运行结果：**

#include <unistd.h>
    #include <stdio.h>
    
    // man 7 mq_overview
    
    int main()
    {
        long mq_open_max = sysconf(_SC_MQ_OPEN_MAX);
        long mq_prio_max = sysconf(_SC_MQ_PRIO_MAX);
        printf("mq_open_max:%ld\n", mq_open_max);
        printf("mq_prio_max:%ld\n", mq_prio_max);
    }

![20160827204405494][]

**高级：**

Posix消息队列允许异步事件通知，以告知合适有一个消息放置到某个空消息队列中。这种通知有两种方式可供选择：

a)产生一个信号

b)创建一个线程来执行一个指定的函数。

这种通知通过调用mq\_notify建立。

MQ_NOTIFY(3)               Linux Programmer's Manual              MQ_NOTIFY(3)
    
    NAME
           mq_notify - register for notification when a message is available
    
    SYNOPSIS
           #include <mqueue.h>
    
           int mq_notify(mqd_t mqdes, const struct sigevent *sevp);
    
           Link with -lrt.

这里不进行深入展开。

参考：《unix网络编程》·卷2

End;

[20160827184413911]: /images/20220717/5299434137064eabadf5eeb3510f29bf.png
[20160827203146610]: /images/20220717/efa4add86db04234916b75a99ea18245.png
[20160827204405494]: /images/20220717/7ce5c731f404418790df07c94bd96dd9.png