《Linux/UNIX系統編程手冊》第52章 POSIX消息隊列

關鍵詞：O_CREAT/O_EXCL、O_NONBLOCK、mq_maxmsg、mq_msgsize、SIGEV_SIGNAL、SIGEV_THREAD等等。

POSIX消息隊列允許進程之間以消息的形式交換數據。POSIX消息隊列和System V消息隊列相似之處在於數據的交換單位都是整個消息。

差別在於：

POSIX消息隊列時引用計數的。只有當所有當前使用隊列的進程都關閉了隊列之后才會對隊列進行標記以便刪除。

每個System V消息都有一個整數類型，並且通過msgrcv()可以以各種方式類型選擇消息。POSIX消息有一個關聯的優先級，並且消息之間是嚴格按照優先級順序排隊的。

POSIX消息隊列提供了一個特性允許在隊列中的一條消息可用於異步地通知進程。

1. 概述

POSIX消息隊列API主要函數如下：

mq_open()：創建一個新消息隊列或打開一個既有隊列，返回后繼調用中會用到的消息隊列描述符。

mq_send()：向對列寫入一條信息。

mq_receive()：從隊列中讀取一條信息。

mq_close()：關閉進程之前打開的一個消息隊列。

mq_unlink()：刪除一個消息隊列名並當所有進程關閉該隊列時對隊列進行標記以便刪除。

mq_getatrr()/mq_setattr()：每個消息隊列都有一組關聯的特性，可以在mq_uoen()創建或者打開隊列時進行設置。獲取和修改隊列特性工作還可以由此兩特性完成。

mq_notify()：允許一個進程向一個隊列注冊接收消息通知。注冊完成后，當一條消息可用時會通過發送一個信號或在一個單獨的線程中調用一個函數來通知進程。

2. 打開、關閉和斷開鏈接消息隊列

mq_open()函數創建一個消息隊列或打開一個既有隊列。

#include <fcntl.h> /* Defines O_* constants */
#include <sys/stat.h> /* Defines 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);
　　Returns a message queue descriptor on success, or (mqd_t) –1 on error

name：標識了消息隊列，必須使用斜線打頭后面跟着一個或多個非斜線字符的名字。Linux和其它一些實現允許采用這種可移植的命名方式來給IPC對象命名。

oflag：是一個位掩碼，具體如下：

oflags其中一個用途是，確定是打開一個既有隊列還是創建和打開一個新隊列。

• 如果oflags中不包含O_CREAT，那么將會打開一個既有隊列；如果隊列不存在，則返回錯誤。
• 如果oflags中包含了O_CREAT，並且給定的name隊列不存在，則創建一個新的空隊列；如果隊列已經存在，則打開此既有隊列。
• 如果oflags中包含了O_CREAT和O_EXCL，並且給定name的隊列已經存在，那么就會失敗；如果不存在則創建一個新的空隊列。

    oflags的O_RDONLY、O_WRONLY、O_RDWR表明調用進程在消息隊列上的訪問方式。

    oflags標記O_NONBLOCK將會導致以非阻塞式模式打開隊列。如果后續的mq_receive()和mq_send()無法在不阻塞情況下執行，那么調用就會立即返回EAGAIN錯誤。

mode：是一個位掩碼，指定了施加於新消息隊列之上的權限。並且與open()一樣，mode中的值會與進程的umask取掩碼。

attr：是一個mq_attr結構，制定了新消息隊列的特性。如果attr為NULL，那么將使用系統定義的默認特性創建隊列。

 

fork()、exec()以及進程終止對消息隊列描述符的影響

在fork()中子進程會接收其父進程的消息隊列描述符的副本，並且這些描述符會引用同樣的打開着的消息隊列描述。

子進程不會繼承其父進程的任何消息通知注冊。

當一個進程執行了一個exec()或終止時，所有其打開的消息隊列描述符會被關閉。關閉消息隊列描述符的結果是進程在相應隊列上的消息通知會被注銷。

 

mq_close()函數關閉消息隊列描述符mqdes。

#include <mqueue.h>
int mq_close(mqd_t mqdes);
　　Returns 0 on success, or –1 on error

如果調用進程已經通過mqdes在隊列上注冊了消息通知，那么通知注冊會自動被刪除，並且另一個進程可以隨后向該隊列注冊消息通知。

當進程終止或調用exec()時，消息隊列描述符會被自動關閉。與文件描述符一陽，應用程序應該在不再使用消息隊列描述符的時候顯式地關閉消息隊列描述符以防止出現進程耗盡消息隊列描述符的情況。

 

mq_unlink()函數刪除通過name標識的消息隊列，並將隊列標記為在所有進程使用完該隊列之后銷毀該隊列。

#include <mqueue.h>
int mq_unlink(const char *name);
    Returns 0 on success, or –1 on error

 mq_close()並不會刪除消息隊列，mq_unlink()才會刪除消息隊列。 

3. 描述符和消息隊列之間的關系

消息隊列描述符和打開着的消息隊列之間的關系與文件描述符和打開着的文件之間的關系類似。

消息隊列描述符是一個進程級別的句柄，它引用了系統層面的打開着的消息隊列描述表中的一個條目，而該條目則引用了一個消息隊列對象。

 

 

• 一個打開的消息隊列描述符擁有一組關聯的標記，即NONBLOCK，它確定了I/O是否是非阻塞的。
• 兩個進程能夠持有引用同一個打開的消息隊列描述的消息隊列描述符(進程A中的x和進程B中的x)。當一個進程在打開了一個消息隊列之后調用fork()時就會發生這種情況。這些描述符會共享O_NONBLOCK標記的狀態。
• 兩個進程能夠持有引用不同消息隊列描述的打開的消息隊列描述(進程A中的z和進程B中的y都引用了/mq-r)。當兩個進程分別使用mq_open()打開同一個隊列時就會發生這種情況。

4. 消息隊列特性

 mq_open()、mq_getattr()以及mq_setattr()都可以設置或者讀取消息隊列特性，通過指向mq_attr結構的指針實現。

struct mq_attr {
　　long mq_flags;   --消息隊列flag[mq_getattr(), mq_setattr()]
　　long mq_maxmsg;  --定義了使用mq_send()向消息隊列上添加消息數量上限，必須大於0。mq_open()時確定，后面無法修改。[mq_open(), mq_getattr()]
　　long mq_msgsize; --定義了加入詳細隊列的每條消息的大小上限，必須大於0。mq_open()時確定，后面無法修改。[mq_open(), mq_getattr()]
　　long mq_curmsgs; --當前消息隊列中消息數目[mq_getattr()]
};

獲取消息隊列特性

mq_getattr()函數返回一個包含與描述符mqdes相關聯的消息隊列的相關信息mq_attr結構。

#include <mqueue.h>
int mq_getattr(mqd_t mqdes, struct mq_attr *attr);
　　Returns 0 on success, or –1 on error

mq_flags取值只有一個O_NONBLOCK，這個標記是mq_open()的oflag參數來初始化的，並且使用mq_setattr()可以修改這個標記。 

修改消息隊列特性

mq_setattr()函數設置與mqdes香瓜年的詳細隊列描述符的特性。

#include <mqueue.h>
int mq_setattr(mqd_t mqdes, const struct mq_attr *newattr, struct mq_attr *oldattr);
　　Returns 0 on success, or –1 on error

使用 newaddr中ma_flags字段來修改mqdes相關聯消息隊列描述的標記。

如果oldattr不為NULL，那么就返回一個之前消息特性的mq_attr結構。

SUSv3規定使用mq_setattr()能夠修改的唯一特性是O_NONBLOCK標記的狀態。

5. 交換信息

5.1 發送消息

mq_send()函數將位於msg_ptr指向的緩沖區中的消息添加到描述符mqdes所引用的消息隊列中。

#include <mqueue.h>
int mq_send(mqd_t mqdes, const char *msg_ptr, size_t msg_len, unsigned int msg_prio);
　　Returns 0 on success, or –1 on error

msg_len參數指定了msg_ptr指向消息的長度，必須小於等於mq_msgsize。否則返回EMSGSIZE錯誤。長度為零是允許的。

msg_prio表示消息的優先級，0表示最低優先級，最大優先級為MQ_PRIO_MAX或者同sysconf(_SC_MQ_PRIO_MAX)返回值來確定。

一條消息被添加到隊列中時，它會被放置在隊列中具有相同優先級的所有消息之后。

如果消息隊列滿，那么后續mq_send()調回會阻塞直到隊列中存在可用空間為止，或者在O_NONBLOCK情況下立即失敗並返回EAGAIN錯誤。

5.2 接收消息

mq_receive()函數從mqdes引用的消息隊列中刪除一條優先級最高、存在時間最長的消息，並將刪除的消息放置在msg_ptr指向的緩沖區。

#include <mqueue.h>
ssize_t mq_receive(mqd_t mqdes, char *msg_ptr, size_t msg_len, unsigned int *msg_prio);
　　Returns number of bytes in received message on success, or –1 on error

msg_len指定msg_ptr指向的換種區中的可用字節數。

msg_len必須大於或等於隊列ms_msgsize，否則mq_receive()就會失敗並返回EMSGSIZE錯誤。

msg_prio不為NULL，那么接收到消息的優先級就會被復制到msg_prio。

如果消息隊列為空，那么mq_receive()會阻塞直到存在可用的消息，或者在O_NONBLOCK情況下會立即失敗並返回EAGAIN。

5.3 設置發送和接收消息超時時間

mq_timedsend()和mq_timedreceive()函數與mq_sned()和mq_receive()基於一樣，唯一的區別在於：如果操作無法立即執行，並且該消息隊列oflag非O_NONBLOCK，那么abs_timeout參數就會為調用阻塞的時間制定一個上限。

#define _XOPEN_SOURCE 600
#include <mqueue.h>
#include <time.h>
int mq_timedsend(mqd_t mqdes, const char *msg_ptr, size_t msg_len, unsigned int msg_prio, const struct timespec *abs_timeout);
　　Returns 0 on success, or –1 on error
ssize_t mq_timedreceive(mqd_t mqdes, char *msg_ptr, size_t msg_len, unsigned int *msg_prio, const struct timespec *abs_timeout);
　　Returns number of bytes in received message on success, or –1 on error

abs_time是一個絕對值，如果要指定一個相對超時可以使用clock_gettime()來獲取CLOCK_REALTIME的當前值並加上所需的時間量來生成一個恰當初始化過側timespec結構。

如果mq_timedsend()或mq_timedreceive()調用因超時而無法完成操作，那么調用就會失敗並返回ETIMEOUT錯誤。

如果abs_timeout為NULL，表示永遠不會超時。

5.4 消息隊列的創建、發送、接收

#include <mqueue.h>
#include <sys/stat.h>
#include <fcntl.h>
#include "tlpi_hdr.h"

static void
usageError(const char *progName)
{
    fprintf(stderr, "Usage: %s [-cx] [-m maxmsg] [-s msgsize] mq-name "
            "[octal-perms]\n", progName);
    fprintf(stderr, "    -c          Create queue (O_CREAT)\n");
    fprintf(stderr, "    -m maxmsg   Set maximum # of messages\n");
    fprintf(stderr, "    -s msgsize  Set maximum message size\n");
    fprintf(stderr, "    -x          Create exclusively (O_EXCL)\n");
    exit(EXIT_FAILURE);
}

int
main(int argc, char *argv[])
{
    int flags, opt;
    mode_t perms;
    mqd_t mqd;
    struct mq_attr attr, *attrp;

    /* If 'attrp' is NULL, mq_open() uses default attributes. If an
       option specifying a message queue attribute is supplied on the
       command line, we save the attribute in 'attr' and set 'attrp'
       pointing to 'attr'. We assign some (arbitrary) default values
       to the fields of 'attr' in case the user specifies the value
       for one of the queue attributes, but not the other. */

    attrp = NULL;
    attr.mq_maxmsg = 10;
    attr.mq_msgsize = 2048;
    flags = O_RDWR;

    /* Parse command-line options */

    while ((opt = getopt(argc, argv, "cm:s:x")) != -1) {
        switch (opt) {
        case 'c':
            flags |= O_CREAT;
            break;

        case 'm':
            attr.mq_maxmsg = atoi(optarg);
            attrp = &attr;
            break;

        case 's':
            attr.mq_msgsize = atoi(optarg);
            attrp = &attr;
            break;

        case 'x':
            flags |= O_EXCL;
            break;

        default:
            usageError(argv[0]);
        }
    }

    if (optind >= argc)
        usageError(argv[0]);

    perms = (argc <= optind + 1) ? (S_IRUSR | S_IWUSR) :
                getInt(argv[optind + 1], GN_BASE_8, "octal-perms");

    mqd = mq_open(argv[optind], flags, perms, attrp);
    if (mqd == (mqd_t) -1)
        errExit("mq_open");

    exit(EXIT_SUCCESS);
}

 

 

#include <mqueue.h>
#include <fcntl.h>              /* For definition of O_NONBLOCK */
#include "tlpi_hdr.h"

static void
usageError(const char *progName)
{
    fprintf(stderr, "Usage: %s [-n] mq-name\n", progName);
    fprintf(stderr, "    -n           Use O_NONBLOCK flag\n");
    exit(EXIT_FAILURE);
}

int
main(int argc, char *argv[])
{
    int flags, opt;
    mqd_t mqd;
    unsigned int prio;
    void *buffer;
    struct mq_attr attr;
    ssize_t numRead;

    flags = O_RDONLY;
    while ((opt = getopt(argc, argv, "n")) != -1) {
        switch (opt) {
        case 'n':   flags |= O_NONBLOCK;        break;
        default:    usageError(argv[0]);
        }
    }

    if (optind >= argc)
        usageError(argv[0]);

    mqd = mq_open(argv[optind], flags);
    if (mqd == (mqd_t) -1)
        errExit("mq_open");

    /* We need to know the 'mq_msgsize' attribute of the queue in
       order to determine the size of the buffer for mq_receive() */

    if (mq_getattr(mqd, &attr) == -1)
        errExit("mq_getattr");

    buffer = malloc(attr.mq_msgsize);
    if (buffer == NULL)
        errExit("malloc");

    numRead = mq_receive(mqd, buffer, attr.mq_msgsize, &prio);
    if (numRead == -1)
        errExit("mq_receive");

    printf("Read %ld bytes; priority = %u\n", (long) numRead, prio);
    if (write(STDOUT_FILENO, buffer, numRead) == -1)
        errExit("write");
    write(STDOUT_FILENO, "\n", 1);

    exit(EXIT_SUCCESS);
}

 

 

#include <mqueue.h>
#include <fcntl.h>              /* For definition of O_NONBLOCK */
#include "tlpi_hdr.h"

static void
usageError(const char *progName)
{
    fprintf(stderr, "Usage: %s [-n] mq-name msg [prio]\n", progName);
    fprintf(stderr, "    -n           Use O_NONBLOCK flag\n");
    exit(EXIT_FAILURE);
}

int
main(int argc, char *argv[])
{
    int flags, opt;
    mqd_t mqd;
    unsigned int prio;

    flags = O_WRONLY;
    while ((opt = getopt(argc, argv, "n")) != -1) {
        switch (opt) {
        case 'n':   flags |= O_NONBLOCK;        break;
        default:    usageError(argv[0]);
        }
    }

    if (optind + 1 >= argc)
        usageError(argv[0]);

    mqd = mq_open(argv[optind], flags);
    if (mqd == (mqd_t) -1)
        errExit("mq_open");

    prio = (argc > optind + 2) ? atoi(argv[optind + 2]) : 0;

    if (mq_send(mqd, argv[optind + 1], strlen(argv[optind + 1]), prio) == -1)
        errExit("mq_send");
    exit(EXIT_SUCCESS);
}

 

 

6. 消息通知

POSIX消息隊列區別於System V消息隊列的一個特性是POSIX消息隊列能夠接收之前為空的隊列上有可用消息的異步通知，即這個隊列從空變成了非控。

這個特性意味着無需執行一個阻塞或者將消息標記為O_NONBLOCK並在隊列上定期執行mq_receive()調用。因為一個進程能夠請求消息到達后通知，然后繼續執行其他任務直到收到通知為止。

進程可以選擇通過信號的形式或創建一個單獨的線程處理函數的形式來接收通知。

• 任一時刻只有一個進程能夠向一個特定的消息隊列注冊接收通知。如果消息隊列上已經存在注冊進程，那么后續在該隊列上的注冊請求將會失敗，mq_notify()返回EBUSY錯誤。
• 只有當一條新消息進入之前為空的隊列時注冊進程才會收到通知。如果在注冊的時候隊列中已經包含消息，那么之后當隊列被清空之后有一條新消息到達之時才會發出通知。
• 當向注冊進程發送了一個通知之后就會刪除注冊信息，之后任何進程就能夠向隊列注冊接收通知了。一個進程要想持續地接收通知，那么他就必須要在每次接收到通知之后再次調用mq_notify()來注冊自己。
• 注冊進程只有在當前不存在其他在該隊列上調用mq_receive()而發生阻塞的進程時才會收到通知。如果其他進程在mq_receive()調用中被阻塞了，那么該進程會讀取消息，注冊進程會保持注冊狀態。
• 一個進程可以通過在調用mq_notify()時傳入一個值為NULL的notification參數來撤銷自己在消息通知上的注冊信息。

mq_notify()函數注冊調用進程在一條消息進入描述符mqdes引用的空隊列時接收通知。

#include <mqueue.h>
int mq_notify(mqd_t mqdes, const struct sigevent *notification);
    Returns 0 on success, or –1 on error

notification參數指定了進程接收通知的機制。

union sigval {
    int sival_int; /* Integer value for accompanying data */
    void *sival_ptr; /* Pointer value for accompanying data */
};

struct sigevent {
    int sigev_notify; /* Notification method */
    int sigev_signo; /* Notification signal for SIGEV_SIGNAL */
    union sigval sigev_value; /* Value passed to signal handler or thread function */
    void (*sigev_notify_function) (union sigval); /* Thread notification function */
    void *sigev_notify_attributes; /* Really 'pthread_attr_t' */
};

其中sigev_notify字段被設置成下列值中的一個：

SIGEV_NONE：注冊這個進程接收通知，但當一條消息進入之前為空的隊列時不通知改進程。與往常一樣，當新消息進入空隊列之后注冊消息會被刪除。

SIGEV_SIGNAL：通過生成一個sigev_signo指定的信號來通知進程。如果sigev_signo是一個實時信號，那么sigev_value字段將會指定信號都帶的數據。通過傳入信號處理器的siginfo_t中的si_value或通過sigwaitinfo()或sigtimedwait()返回值都能夠去的這部分數據。

typedef struct {
　　int si_signo; /* Signal number */
　　int si_code; /* Signal code */----------對應應該為SI_MESGQ
　　int si_trapno; /* Trap number for hardware-generated signal (unused on most architectures) */
　　union sigval si_value; /* Accompanying data from sigqueue() */
　　pid_t si_pid; /* Process ID of sending process */
　　uid_t si_uid; /* Real user ID of sender */
...
} siginfo_t;

SIGEV_THREAD：通過調用在sigev_notify_function中指定的函數來通知進程，就像是在一個新線程中啟動該函數一樣。sigev_notify_attreibutes字段你可以為NULL或是一個指向定義了線程特性的pthread_attr_t指針。sigev_value中指定的sigval值將作為參數傳入這個參數。

6.1 通過信號接收通知

通過sigaction()注冊信號處理函數，通過mq_notify()在消息隊列上注冊通知，當空隊列有新消息到達后，會產生一個信號。

系統會執行信號處理函數，然后循環讀取隊列知道清空；在for()中循環注冊通知，然后讀取消息。

#include <signal.h>
#include <mqueue.h>
#include <fcntl.h>              /* For definition of O_NONBLOCK */
#include "tlpi_hdr.h"

#define NOTIFY_SIG SIGUSR1

static void handler(int sig)
{
    /* Just interrupt sigsuspend() */
    printf("New message coming.\n");
}

int main(int argc, char *argv[])
{
    struct sigevent sev;
    mqd_t mqd;
    struct mq_attr attr;
    void *buffer;
    ssize_t numRead;
    sigset_t blockMask, emptyMask;
    struct sigaction sa;

    if (argc != 2 || strcmp(argv[1], "--help") == 0)
        usageErr("%s mq-name\n", argv[0]);

    mqd = mq_open(argv[1], O_RDONLY | O_NONBLOCK);------------以O_NONBLOCK的方式打開消息隊列。 if (mqd == (mqd_t) -1)
        errExit("mq_open");

    /* Determine mq_msgsize for message queue, and allocate an input buffer
       of that size */

    if (mq_getattr(mqd, &attr) == -1)--------從消息隊列獲取mq_msgsize特性值。
        errExit("mq_getattr");

    buffer = malloc(attr.mq_msgsize);--------根據mq_msgsize的值申請接收消息緩沖區。 if (buffer == NULL)
        errExit("malloc");

    /* Block the notification signal and establish a handler for it */

    sigemptyset(&blockMask);
    sigaddset(&blockMask, NOTIFY_SIG);
    if (sigprocmask(SIG_BLOCK, &blockMask, NULL) == -1)------阻塞SIGUSR1通知信號。
        errExit("sigprocmask");

    sigemptyset(&sa.sa_mask);
    sa.sa_flags = 0;
    sa.sa_handler = handler;
    if (sigaction(NOTIFY_SIG, &sa, NULL) == -1)--------------給SIGUSR1通知信號建立處理器handler()。
        errExit("sigaction");

    /* Register for message notification via a signal */

    sev.sigev_notify = SIGEV_SIGNAL;
    sev.sigev_signo = NOTIFY_SIG;
    if (mq_notify(mqd, &sev) == -1)--------------------------調用mq_notify()來注冊進程消息通知。
        errExit("mq_notify");

    sigemptyset(&emptyMask);

    for (;;) {
        sigsuspend(&emptyMask);--------------------在此解除通知信號的阻塞狀態，並等待；直到消息隊列變為空情況下有新消息到達，產生信號后被捕獲。然后執行handler()處理器函數。同時進程的消息通知注冊信息會被刪除。
                               --------------------這里使用sigsuspend()而不是pause()，是為了防止出現程序在執行for中其他代碼是錯誤信號的情況。
        if (mq_notify(mqd, &sev) == -1)------------重新注冊進程接收消息通知。如果不重新注冊，那么將永遠不會產生。
            errExit("mq_notify");
                                       ------------使用while循環讀取隊列中所有消息，可以清空隊列能夠確保當一條新消息到達之后會產生一個新通知。 while ((numRead = mq_receive(mqd, buffer, attr.mq_msgsize, NULL)) >= 0)-----從當前消息隊列中讀取消息。
            printf("Read %ld bytes\n", (long) numRead);

        if (errno != EAGAIN)            /* Unexpected error */------------因為O_NONBLOCK的方式打開，所以在消息隊列被清空之后就會終止，mq_receive()會失敗並返回EAGAIN。
            errExit("mq_receive");
    }
}

 

從實際使用看，O_NONBLOCK+信號處理的方式要比以阻塞打開，創建一個線程並在線程中進行mq_receive()復雜。

1. mq_notify()生效時消息隊列需要為空。

2. 需要重復注冊mq_notify()。

6.2 通過線程接收通知

通過mq_notify()注冊線程消息通知，在空隊列有消息到達后創建一個線程執行線程函數。

#include <pthread.h>
#include <mqueue.h>
#include <fcntl.h>              /* For definition of O_NONBLOCK */
#include "tlpi_hdr.h"

static void notifySetup(mqd_t *mqdp);

static void threadFunc(union sigval sv)
{
    ssize_t numRead;
    mqd_t *mqdp;
    void *buffer;
    struct mq_attr attr;

    mqdp = sv.sival_ptr;

    if (mq_getattr(*mqdp, &attr) == -1)
        errExit("mq_getattr");

    buffer = malloc(attr.mq_msgsize);------------根據消息隊列屬性申請內存。 if (buffer == NULL)
        errExit("malloc");

    /* Reregister for message notification */

    notifySetup(mqdp);---------------必須在清空消息隊列之前，重新注冊消息通知。 while ((numRead = mq_receive(*mqdp, buffer, attr.mq_msgsize, NULL)) >= 0)-----一次性讀取完所有消息，清空隊列。
        printf("Read %ld bytes\n", (long) numRead);

    if (errno != EAGAIN)                        /* Unexpected error */
        errExit("mq_receive");

    free(buffer);
}

static void
notifySetup(mqd_t *mqdp)
{
    struct sigevent sev;

    sev.sigev_notify = SIGEV_THREAD;            /* Notify via thread */
    sev.sigev_notify_function = threadFunc;----------------設置消息通知方式為啟動線程，並且線程處理函數為threadFunc()。
    sev.sigev_notify_attributes = NULL;
            /* Could be pointer to pthread_attr_t structure */
    sev.sigev_value.sival_ptr = mqdp;      ----------------確保在threadFunc中可以調用mq_getattr()/mq_receive()等。

    if (mq_notify(*mqdp, &sev) == -1)
        errExit("mq_notify");
}

int
main(int argc, char *argv[])
{
    mqd_t mqd;

    if (argc != 2 || strcmp(argv[1], "--help") == 0)
        usageErr("%s mq-name\n", argv[0]);

    mqd = mq_open(argv[1], O_RDONLY | O_NONBLOCK);
    if (mqd == (mqd_t) -1)
        errExit("mq_open");

    notifySetup(&mqd);
    pause(); -----------------主程序永遠暫停在此。
}

7. Linux特有的特性

可以通過mount將POSIX消息隊列掛載到文件系統中：

mount -t mqueue none /dev/mqueue

在/dev/mqueue中，可以查看每個mqueue的屬性：

cat /dev/mqueue/mq 
QSIZE:0          NOTIFY:0     SIGNO:0     NOTIFY_PID:0

QSIZE：為隊列中所有數據的總字節數。

NOTIFY：與sigev_notify對應的值，0表示SIGEV_SIGNAL，1表示SIGEV_NONE，2表示SIGEV_THREAD。

SIGNO：為使用哪個信號來分發消息通知，0表示沒有采用信號處理。

NOTIFY_PID：為向該隊列注冊通知的進程ID，0表示沒有注冊。

8. 消息隊列限制

MQ_PRIO_MAX：定義一條消息的最大優先級。

MQ_OPEN_MAX：限制一個進程最多能打開的消息隊列數量。

msg_max：限制為新消息隊列的mq_maxmsg特性的取值規定了一個上限。

msgsize_max：限制為非特權進程創建的新消息隊列的mq_msgsize特性的取值規定了一個上限。當一個特權進程調用mq_open()時會忽略這個限制。

queues_max：是一個系統級別的限制，規定了系統上最多能夠創建的消息隊列的數量。一旦達到這個限制，只有特權進程才能夠創建新都列。

9. POSIX和System V消息隊列比較

POSIX消息隊列相對於System V消息隊列優勢：

1. POSIX IPC接口更加簡單，並與UNIX文件模型更加一致。

2. POSIX IPC對象是引用技術的，簡化了確定合適刪除一個對象的任務。

3. 消息通知特性允許一個進程能夠在一條消息進入之前為空的隊列時異步地通過信號或線程的實例化來接收通知。

4. 在Linux上可以使用poll()、select()以及epoll來監控POSIX消息隊列。

POSIX消息隊列相對劣勢：

1. POSIX消息隊列可移植性稍差。

2. 與POSIX消息隊列嚴格按照優先級排序相比，System V消息隊列能夠根據類型來選擇消息的功能更加靈活。