基本概念：

    訊息佇列可認為是一個訊息連結串列。有足夠寫許可權的執行緒可往佇列中放置訊息，有足夠讀許可權的執行緒可從佇列中取走訊息，每個訊息都是一個記錄（非位元組流式，也就是不需要自定義邊界），它由傳送者賦予一個優先順序。在某個程序往一個佇列寫入訊息之前，並不需要另外某個程序在該佇列上等待訊息的到達。

    一個程序可以往某個佇列寫入一些訊息，然後終止，再讓另外一個程序在以後某個時刻讀出這些訊息。訊息佇列具有隨核心的持續性，也就是程序關閉後，訊息佇列依然存在。

訊息佇列的標準：

    訊息佇列分為兩大標準，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;
}

編譯執行：

傳送訊息：

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;
}

編譯執行：

訊息佇列限制：

我們已經遇到任意給定佇列的兩個限制，它們都是在建立該佇列時建立的（測試中發現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);
}

高階：

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;