Unix/Linux编程：POSIX 消息队列

女爷i 2022-10-05 03:59 361阅读 0赞

POSIX 消息队列允许进程之间以消息的形式交换数据。POSIX消息队列与System V消息队列的相似之处在于数据的交换单位是整个消息，但它们之间存在一些显著的差异

*  POSIX消息队列是引用计数的。只有当所有当前使用队列的进程都关闭了之后才会对队列进程标记以便删除
 *  每个System V消息都有一个整数类型，并且通过msgrcv()可以以各种方式选择消息。而Posix 消息由一个管理的优先级，并且消息之间是严格按照优先级顺序排队(以及接收)的
 *  POSIX消息队列提供了一个特性允许在队列中的一条消息可用时异步的通知进程
 *  POSIX消息通知特性运行一个进程能够在一条消息进入空队列时异步通知信号或者线程的实例化来接受通知
 *  在Linux上可以使用poll、select、epoll来监听POSIX消息队列。System消息没有这个特性

缺点：

*  POSIX 消息队列的可移植性稍差
 *  与 POSIX 消息队列严格按照优先级排序相比，System V 消息队列能够根据类型来选择消息的功能的灵活性更强。

POSIX 消息队列支持是一个通过 CONFIG\_POSIX\_MQUEUE 选项配置的可选内核组件。

# 概述 #

POSIX 消息队列API的主要函数如下

*  mq\_open()函数创建一个新消息队列或打开一个既有队列，返回后续调用中会用到的消息队列描述符。
 *  mq\_send()函数向队列写入一条消息
 *  mq\_receive()函数从队列中读取一条消息。
 *  mq\_close()函数关闭进程之前打开的一个消息队列。
 *  mq\_unlink()函数删除一个消息队列并当所有进程关闭该队列时对队列进程标记以便删除

此外，POSIX消息队列API还具备一些特别的特性：

*  每个消息队列都有一组关联的特性，其中一些特性可以在使用mq\_open()创建或者打开时进行设置。获取和修改队列特性的工作是由mq\_getattr()和mq\_setaddr()来完成的
 *  mq\_notify()函数允许一个进程向一个队列注册时接收消息通知。在注册完之后，当一条消息可用时会通过发送一个信号或者在一个单独的线程中调用一个函数来通知进程

## 打开一个消息队列 ##

mq\_open()函数创建一个新消息队列或打开一个既有队列。

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.
    
    RETURN VALUE
           成功时，mq_open() 返回一个消息队列描述符供其他消息队列函数使用。 
           出错时，mq_open() 返回 (mqd_t) -1，设置 errno 以指示错误

name 参数标识出了消息队列

oflag 参数是一个位掩码，它控制着 mq\_open()操作的各个方面，取值如下：

<table> 
 <thead> 
  <tr> 
   <th>标记</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>O_CREAT</td> 
   <td>队列不存在时创建队列</td> 
  </tr> 
  <tr> 
   <td>O_EXCL</td> 
   <td>与 O_CREAT 一起排它地创建队列</td> 
  </tr> 
  <tr> 
   <td>O_RDONLY</td> 
   <td>只读打开</td> 
  </tr> 
  <tr> 
   <td>O_WRONLY</td> 
   <td>只写打开</td> 
  </tr> 
  <tr> 
   <td>O_RDWR</td> 
   <td>读写打开</td> 
  </tr> 
  <tr> 
   <td>O_NONBLOCK</td> 
   <td>以非阻塞模式打开</td> 
  </tr> 
 </tbody> 
</table>

O\_NONBLOCK 将会导致以非阻塞的模式打开队列。如果后续的mq\_receive()或 mq\_send()调用无法在不阻塞的情况下执行，那么调用就会立即返回 EAGAIN 错误

mq\_open()通常用来打开一个既有消息队列，这种调用只需要两个参数，但如果在 flags中指定了 O\_CREAT，那么就还需要另外两个参数：mode 和 attr。（如果通过 name 指定的队列已经存在，那么这两个参数会被忽略。）这些参数的用法如下。

*  mode 参数是一个位掩码，它指定了施加于新消息队列之上的权限
 *  attr 参数是一个 mq\_attr 结构，它指定了新消息队列的特性。如果 attr 为 NULL，那么将使用实现定义的默认特性创建队列

mq\_open()在成功结束时会返回一个消息队列描述符，它是一个类型为 mqd\_t 的值，在后续的调用中将会使用它来引用这个打开着的消息队列。SUSv3 对这个数据类型的唯一约束是它不能是一个数组，即需要确保这个类型是一个能在赋值语句中使用或者是能够作为函数参数返回的类型。（如在 Linux 上，mqd\_t 是一个 int，而在 Solaris 上将其定义为 void \*。

> fork()、exec()以及进程终止对消息队列描述符的影响

在fork()中子进程会接受其父进程在消息队列描述符的副本，并且这些描述符会引用同样的打开的消息队列描述。子进程不会继承父进程的任何消息通知注册。

当一个进程执行了一个 exec()或终止时，所有其打开的消息队列描述符会被关闭。关闭消息队列描述符的结果是进程在相应队列上的消息通知注册会被注销。

// pmsg_create.c
    #include <unistd.h>
    #include <signal.h>
    #include <stdio.h>
    #include <malloc.h>
    #include <stdlib.h>
    #include <errno.h>
    #include <sys/mman.h>
    #include <string.h>
    #include <semaphore.h>
    #include <sys/stat.h>
    #include <fcntl.h>
    #include <mqueue.h>
    #include <sys/stat.h>
    #include <fcntl.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) :
                atoi(argv[optind + 1]);
    
        mqd = mq_open(argv[optind], flags, perms, attrp);
        if (mqd == (mqd_t) -1){
        
            printf("mq_open");
            exit(EXIT_FAILURE);
        }
            
    
        exit(EXIT_SUCCESS);
    }

## 关闭一个消息队列 ##

NAME
           mq_close - close a message queue descriptor
    
    SYNOPSIS
           #include <mqueue.h>
    
           int mq_close(mqd_t mqdes);
    
           Link with -lrt.

如果调用进程已经通过mqdes在队列上注册了消息通知，那么通知注册会自动被删除，并且另一个进程可以随后向该队列注册消息通知。

当进程终止或者调用exec()时，消息队列描述符会被自动关闭。问问价描述符一样，应用程序应该在不再适用消息队列描述符的时候显示的关闭消息队列描述符以防止进程耗尽消息队列描述符的情况。

与文件上的close()一样，关闭一个消息队列并不会删除该队列。要删除队列必须显示的调用mq\_unlink()，它是unlink()在消息队列上的版本。

## 删除一个消息队列 ##

mq\_unlink()函数删除通过 name 标识的消息队列，并将队列标记为在所有进程使用完该队列之后销毁该队列

NAME
           mq_unlink - remove a message queue
    
    SYNOPSIS
           #include <mqueue.h>
    
           int mq_unlink(const char *name);
    
           Link with -lrt.

下面是用法：

// 使用 mq_unlink()断开一个 POSIX 消息队列的链接
    int main(int argc, char *argv[])
    {
        
        if (argc != 2 || strcmp(argv[1], "--help") == 0){
        
    		 printf("%s mq-name\n", argv[0]);
    		 exit(EXIT_FAILURE);
    	}
            
    
        if (mq_unlink(argv[1]) == -1){
        
        	 perror("mq_unlink"); 
    		 exit(EXIT_FAILURE);
    	}
           
        exit(EXIT_SUCCESS);
    }

# 描述符和消息队列之间的关系 #

消息队列是一个进程级别的句柄，它引用了系统层面的打开的消息队列描述符中的一个条目，而该条目引用了一个消息队列对象，如下图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXpoZW5nZ3Vhbg_size_16_color_FFFFFF_t_70]

*  一个打开的消息队列描述拥有一组关联的标记。SUSv3只规定了一种这样的标记，  
    NONBLOCK，它确定了 I/O 是否是非阻塞的。  
    两个进程能够持有引用同一个打开的消息队列描述的消息队列描述符(上图中的x)。当一个进程在打开了一个消息队列之后调用fork()时就会发生这样的情况。这些描述符会共享O\_NONBLOCK标记的状态
 *  两个进程能够持有引用不同消息队列描述(它们引用了同一个消息队列)的打开的消息队列描述（如进程 A 中的描述符 z 和进程 B 中的描述符 y 都引用了/mq-r）。当两个进程分别使用 mq\_open()打开同一个队列时就会发生这种情况

# 消息队列特性 #

mq\_open()、mq\_getattr()以及 mq\_setattr()函数都会接收一个参数，它是一个指向 mq\_attr结构的指针。这个结构是在<mqueue.h>中进行定义的，其形式如下。

struct mq_attr
    {
        
      __syscall_slong_t mq_flags;	/* Message queue flags.  */
      __syscall_slong_t mq_maxmsg;	/* Maximum number of messages.  */
      __syscall_slong_t mq_msgsize;	/* Maximum message size.  */
      __syscall_slong_t mq_curmsgs;	/* Number of messages currently queued.  */
      __syscall_slong_t __pad[4];
    };

## 在创建队列时设置消息队列特性 ##

在使用 mq\_open()创建消息队列时可以通过下列 mq\_attr 字段来确定队列的特性

*  mq\_maxmsg 字段定义了使用mq\_send()向消息队列添加消息的数量上限，其取值必须大于0
 *  mq\_msgsize 字段定义了加入消息队列的每条消息的大小的上限，其取值必须大于 0

内核根据这两个值来确定消息队列所需的最大内存量。

mq\_maxmsg和mq\_msgsize特性是在消息队列被创建时就确定下来的，并且之后也无法修改这两个特性。

下面程序显示了如何创建一个 POSIX 消息队列：消息队列特性可以通过两个命令行参数来指定：–m 用于指定 mq\_maxmsg，–s 用于指定mq\_msgsize。只要指定了其中一个选项，那么一个非 NULL 的 attrp 参数就会被传递给mq\_open()。如果在命令行中只指定了–m 和–s 选项中的一个，那么 attrp 指向的 mq\_attr 结构中的一些字段就会取默认值。如果两个选项都被没有被指定，那么在调用 mq\_open()时会将attrp 指定为 NULL，这将会导致使用由实现定义的队列特性的默认值来创建队列

## 获取/修改消息队列特性 ##

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, const struct mq_attr *newattr,
                            struct mq_attr *oldattr);
    
           Link with -lrt.

mq\_getattr()函数返回一个包含与描述符 mqdes 相关联的消息队列描述和消息队列的相关信息的 mq\_attr 结构。

除了上面已经介绍的 mq\_maxmsg 和 mq\_msgsize 字段之外，attr 指向的返回结构中还包含下列字段。

*  mq\_flags ：其取值只有一个：O\_NONBLOCK。这个标记是根据 mq\_open()的 oflag 参数来初始化的，并且使用 mq\_setattr()可以修改这个标记。
 *  mq\_curmsgs：这个当前位于队列中的消息数。这个信息在 mq\_getattr()返回时可能已经发生了改变，前提是存在其他进程从队列中读取消息或向队列写入消息

mq\_setattr()函数设置与消息队列描述符 mqdes 相关联的消息队列描述的特性，并可选地返回与消息队列有关的信息。mq\_setattr()函数执行下列任务。

*  它使用 newattr 指向的 mq\_attr 结构中的 mq\_flags 字段来修改与描述符 mqdes 相关联的消息队列描述的标记
 *  如果 oldattr 不为 NULL，那么就返回一个包含之前的消息队列描述标记和消息队列特性的 mq\_attr 结构（即与 mq\_getattr()执行的任务一样）。

SUSv3 规定使用 mq\_setattr()能够修改的唯一特性是O\_NONBLOCK 标记的状态。如下代码可以启用O\_NONBLOCK

if(mq_getattr(mqd, &attr) == -1){
        
    	exit(EXIT_FAILURE);
    }
    attr.mq_flags != O_NONBLOCK;
    if(mq_setattr(mqd, &attr, NULL) == -1){
        
    	exit(EXIT_FAILURE);
    }

下面程序是使用了mq\_getattr()来获取通过命令行参数指定的消息队列的特性，然后在标准输出中显示这些特性。

// 获取 POSIX 消息队列特性 pmsg_getattr.c
    #include <stdio.h>
    #include <string.h>
    #include <stdlib.h>
    #include <mqueue.h>
    #include <sys/stat.h>
    #include <fcntl.h>
    #include <getopt.h>
    
    int
    main(int argc, char *argv[])
    {
        
        mqd_t mqd;
        struct mq_attr attr;
    
        if (argc != 2 || strcmp(argv[1], "--help") == 0){
        
            printf("%s mq-name\n", argv[0]);
            exit(EXIT_FAILURE);
        }
    
    
        mqd = mq_open(argv[1], O_RDONLY);
        if (mqd == (mqd_t) -1){
        
            perror("mq_open");
            exit(EXIT_FAILURE);
        }
    
    
        if (mq_getattr(mqd, &attr) == -1){
        
            perror("mq_getattr");
            exit(EXIT_FAILURE);
        }
    
    
        printf("Maximum # of messages on queue:   %ld\n", attr.mq_maxmsg);
        printf("Maximum message size:             %ld\n", attr.mq_msgsize);
        printf("# of messages currently on queue: %ld\n", attr.mq_curmsgs);
        exit(EXIT_SUCCESS);
    }

![在这里插入图片描述][20210606201618800.png]  
从上面的输出中可以看出 Linux 上 mq\_maxmsg 和 mq\_msgsize 的默认取值分别为 10 和8192

# 交换消息 #

## 发送消息 ##

NAME
           mq_send - 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 int msg_prio);
    
           Link with -lrt.

mq\_send()函数将位于msg\_ptr指向的缓冲区中的消息添加到描述符mqdes所引用的消息队列中

*  msg\_len参数指定了msg\_ptr指向的消息的长度，其值必须小于或者等于队列的mq\_msgsize特性，否则mq\_send()就会返回EMSGSIZE 错误。长度为零的消息是允许的
 *  每个队列都拥有一个用非负整数表示的优先级，它通过msq\_prio指定。0表示优先级最低，数值越大优先级越高。当一个消息被添加到队列中时，他会被放置在队列中具有相同优先级的所有消息之后。如果一个应用程序无需使用消息优先级，那么只需要将msg\_prio 指定为 0 即可。
 *  SUSv3 允许一个实现为消息优先级规定一个上限，这可以通过定义常量 MQ\_PRIO\_MAX或通过规定 sysconf(\_SC\_MQ\_PRIO\_MAX)的返回值来完成。
 *  如果消息队列已经满了（即已经达到了队列的 mq\_maxmsg 限制），那么后续的 mq\_send()调用会阻塞直到队列中存在可用空间为止或者在 O\_NONBLOCK 标记起作用时立即失败并返回 EAGAIN 错误。

下面程序演示了如何向 POSIX 消息队列写入一条消息

// pmsg_send.c
    #include <stdio.h>
    #include <string.h>
    #include <stdlib.h>
    #include <mqueue.h>
    #include <sys/stat.h>
    #include <fcntl.h>
    #include <getopt.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){
        
            perror("mq_open");
            exit(EXIT_FAILURE);
        }
    
    
        prio = (argc > optind + 2) ? atoi(argv[optind + 2]) : 0;
    
        if (mq_send(mqd, argv[optind + 1], strlen(argv[optind + 1]), prio) == -1){
        
            perror("mq_send");
            exit(EXIT_FAILURE);
        }
    
        exit(EXIT_SUCCESS);
    }

## 接收消息 ##

mq\_receive()从mqdes引用的消息队列中删除一条优先级最高、存在时间最长的消息并将删除的消息放置在msq\_ptr指向对的缓冲区

NAME
           mq_receive - 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 int *msg_prio);

调用者通过msq\_len指定msq\_prt指向的缓冲区中可用字节数。

不管消息的实际大小是什么，msg\_len（即 msg\_ptr 指向的缓冲区的大小）必须要大于或等于队列的 mq\_msgsize 特性，否则 mq\_receive()就会失败并返回 EMSGSIZE 错误。

如果 msg\_prio 不为 NULL，那么接收到的消息的优先级会被复制到 msg\_prio 指向的位置处。

如果消息队列当前为空，那么 mq\_receive() 会阻塞直到存在可用的消息或在O\_NONBLOCK 标记起作用时会立即失败并返回 EAGAIN 错误。

下面程序演示了如何从 POSIX 消息队列中读取一条消息

// pmsg_receive.c
    #include <stdio.h>
    #include <string.h>
    #include <stdlib.h>
    #include <mqueue.h>
    #include <sys/stat.h>
    #include <fcntl.h>
    #include <getopt.h>
    #include <zconf.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){
        
            perror("mq_open");
            exit(EXIT_FAILURE);
        }
    
    
        /* 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){
        
            perror("mq_getattr");
            exit(EXIT_FAILURE);
        }
    
    
        buffer = malloc(attr.mq_msgsize);
        if (buffer == NULL){
        
            perror("malloc");
            exit(EXIT_FAILURE);
        }
    
    
        numRead = mq_receive(mqd, (char *)buffer, attr.mq_msgsize, &prio);
        if (numRead == -1){
        
            perror("mq_receive");
            exit(EXIT_FAILURE);
        }
    
    
        printf("Read %ld bytes; priority = %u\n", (long) numRead, prio);
        /*FIXME: above: should use %zd here, and remove (long) cast */
        if (write(STDOUT_FILENO, buffer, numRead) == -1){
        
            perror("write");
            exit(EXIT_FAILURE);
        }
    
        write(STDOUT_FILENO, "\n", 1);
    
        exit(EXIT_SUCCESS);
    }

效果：首先创建了一个消息队列并向其发送了一些具备不同优先级的消息。  
![在这里插入图片描述][20210606203935478.png]  
然后执行一系列命令来从队列中接收消息  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXpoZW5nZ3Vhbg_size_16_color_FFFFFF_t_70 1]  
从上面的输出中可以看出，消息的读取是按照优先级来进行的。  
此刻，这个队列是空的。当再次执行阻塞式接收时，操作就会阻塞  
![在这里插入图片描述][20210606204040504.png]  
另一方面，如果执行了一个非阻塞接收，那么调用就会立即返回一个失败状态。  
![在这里插入图片描述][2021060620405587.png]

## 在发送和接收消息时设置超时时间 ##

#include <time.h>
           #include <mqueue.h>
    
           int mq_timedsend(mqd_t mqdes, const char *msg_ptr,
                         size_t msg_len, unsigned int msg_prio,
                         const struct timespec *abs_timeout);
    
           #include <time.h>
           #include <mqueue.h>
    
           ssize_t mq_timedreceive(mqd_t mqdes, char *msg_ptr,
                              size_t msg_len, unsigned int *msg_prio,
                              const struct timespec *abs_timeout);

mq\_timedsend()和 mq\_timedreceive()函数与 mq\_send()和 mq\_receive()几乎是完全一样的，它们之间唯一的差别在于如果操作无法立即被执行，并且该消息队列描述上的O\_NONBLOCK 标记不起作用，那么 abs\_timeout 参数就会为调用阻塞的时间指定一个上限。

如果 mq\_timedsend()或 mq\_timedreceive()调用因超时而无法完成操作，那么调用就会失败并返回 ETIMEDOUT 错误。

在 Linux 上将 abs\_timeout 指定为 NULL 表示永远不会超时，但这种行为并没有在 SUSv3中得到规定，因此可移植的应用程序不应该依赖这种行为

# 消息通知 #

POSIX消息队列区别于System V消息队列的一个特性是POSIX消息队列能够接收之前为空的队列上有可用消息的异步通知（即队列从空变成了非空）。这个特性意味着已经无需指向一个阻塞的调用或者将消息队列描述符标记为非阻塞并在队列上定期指向mq\_receive消息了，进程可以选择通过信号的形式或者通过在一个单独的线程中调用一个函数的形式来接受通知。

mq\_notify()函数注册调用进程在一条消息进入描述符mqdes引用的空队列时接收通知：

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.

sevp参数指定了进程接受通知的机制。有关消息通知，注意：

*  在任意时刻都只有一个进程(“注册进程”)能够向一个特定的消息队列注册接收通知。如果一个消息队列上已经存在注册线程了，那么后续在该队列上的注册请求将会失败（mq\_notify()返回 EBUSY 错误）。
 *  只有当一条新消息进入空队列时才会发送通知。如果队列不空就不会发送通知消息
 *  当向注册进程发送一个通知之后就会删除注册消息，之后任何进程就能够向队列注册接收通知了。换句话说，只要一个进程想要持续的接收通知，那么它就必须要在每次接收通知之后再次调用mq\_notify注册自己
 *  注册进程只有在当前不存在其他在该队列上调用mq\_receive()而发生阻塞的进程时才会收到通知。如果其他进程在 mq\_receive()调用中被阻塞了，那么该进程会读取消息，注册进程会保持注册状态
 *  一个进程可以通过在调用 mq\_notify()时传入一个值为 NULL 的 notification 参数来撤销自己在消息通知上的注册信息

sigevent 中与mq\_notify()相关的字段：

union sigval{
        
    	int sival_int;
    	void *sival_ptr;
    };
    
    struct sigevent{
        
    	int sigev_notify;
    	int sigev_signo;
    	union sigval sifev_value;
    	void (*sigev_notify_function)(union sigval);
    	void *sigev_notify_attributes;
    };

这个结构的 sigev\_notify 字段将会被设置成下列值中的一个

*  SIGEV\_NONE：注册这个进程接收通知，但当一条消息进入空队列时不通知该进程。与往常一样，当新消息进入空队列之后注册信息会被删除。
 *  SIGEV\_SIGNAL：
    
     *  通过生成一个在sigev\_signo字段中指定的信号来通知进程。
     *  如果sigev\_signo是一个实时信号，那么sigev\_value字段将会指定信号都带的数据。通过传入信号处理器的siginfo\_t 结构中的 si\_value 字段或通过调用 sigwaitinfo()或 sigtimedwait()返回值能够取得这部分数据。
     *  siginfo\_t 结构中的下列字段也会被填充：si\_code，其值为 SI\_MESGQ；si\_signo，其值是信号编号；si\_pid，其值是发送消息的进程的进程 ID；以及 si\_uid，其值是发送消息的进程的真实用户 ID。（si\_pid 和 si\_uid 字段在其他大多数实现上不会被设置。
 *  SIGEV\_THREAD：
    
     *  通过调用在 sigev\_notify\_function 中指定的函数来通知进程，就像是在一个新线程中启动该函数一样。
     *  sigev\_notify\_attributes 字段可以为 NULL 或是一个指向定义了线程的特性的 pthread\_ attr\_t 结构的指针。sigev\_value 中指定的联合 sigval 值将会作为参数传入这个函数。

## 通过信号接收通知 ##

// mq_notify_sig.c
    #include <stdio.h>
    #include <string.h>
    #include <stdlib.h>
    #include <mqueue.h>
    #include <sys/stat.h>
    #include <fcntl.h>
    #include <getopt.h>
    #include <zconf.h>
    #include <signal.h>
    #include <mqueue.h>
    #include <fcntl.h>
    #include <errno.h>
    
    #define NOTIFY_SIG SIGUSR1
    static void handler(int sig)
    {
        
        /* Just interrupt sigsuspend() */
    }
    /* This program does not handle the case where a message already exists on
       the queue by the time the first attempt is made to register for message
       notification. In that case, the program would never receive a notification.
       See mq_notify_via_signal.c for an example of how to deal with that case. */
    
    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){
        
            printf("%s mq-name\n", argv[0]);
            exit(EXIT_FAILURE);
        }
    
        // 以非阻塞模式打开了一个通过命令行指定名称的消息队列
        mqd = mq_open(argv[1], O_RDONLY | O_NONBLOCK);
        if (mqd == (mqd_t) -1){
        
            perror("mq_open");
            exit(EXIT_FAILURE);
        }
    
    
    
        // 确定该队列的mq_msgsize 特性的值
        if (mq_getattr(mqd, &attr) == -1){
        
            perror("mq_getattr");
            exit(EXIT_FAILURE);
        }
    
    	// 分配一个大小为该值的缓冲区来接收消息
        buffer = malloc(attr.mq_msgsize);
        if (buffer == NULL){
        
            perror("malloc");
            exit(EXIT_FAILURE);
        }
    
    
    	// 阻塞通知信号（SIGUSR1）并为其建立一个处理器
        sigemptyset(&blockMask);
        sigaddset(&blockMask, NOTIFY_SIG);
        if (sigprocmask(SIG_BLOCK, &blockMask, NULL) == -1){
        
            perror("sigprocmask");
            exit(EXIT_FAILURE);
        }
    
    
        sigemptyset(&sa.sa_mask);
        sa.sa_flags = 0;
        sa.sa_handler = handler;
        if (sigaction(NOTIFY_SIG, &sa, NULL) == -1){
        
            perror("sigaction");
            exit(EXIT_FAILURE);
        }
    
    	// 首次调用 mq_notify()来注册进程接收消息通知
        sev.sigev_notify = SIGEV_SIGNAL;
        sev.sigev_signo = NOTIFY_SIG;
        if (mq_notify(mqd, &sev) == -1){
        
            perror("mq_notify");
            exit(EXIT_FAILURE);
        }
    
    
        sigemptyset(&emptyMask);
    
        for (;;) {
        
           //sigsuspend()解除通知信号的阻塞状态并等待直到信号被捕获。从这个系统调用
           //中返回表示已经发生了一个消息通知。此刻，进程会撤销消息通知的注册信息。
            sigsuspend(&emptyMask);         /* Wait for notification signal */
    
            /* Reregister for message notification */
    		// 调用 mq_notify()重新注册进程接收消息通知
            if (mq_notify(mqd, &sev) == -1){
        
                perror("mq_notify");
                exit(EXIT_FAILURE);
            }
    
    		//执行一个 while 循环从队列中尽可能多地读取消息以便清空队列
            while ((numRead = mq_receive(mqd, (char *)buffer, attr.mq_msgsize, NULL)) >= 0)
                printf("Read %ld bytes\n", (long) numRead);
            /*FIXME: above: should use %zd here, and remove (long) cast */
    
            if (errno != EAGAIN)            /* Unexpected error */
            {
        
                perror("mq_receive");
                exit(EXIT_FAILURE);
            }
        }
    }

*  程序阻塞了通知信号并使用siigsuspend()来等待该信号，而没有使用pause（），这是为了防止出现程序在执行for循环中的其他代码时错过信号。
 *  程序以非阻塞模式打开了队列，并且当一个通知发送之后使用一个while信号来读取队列中的所有消息。通过这种方式来清空队列可以确保当一条新消息到达之后会产生一个新通知。使用非阻塞模式意味着while循环在队列中被清空之后就会终止（mq\_receive()会失败并返回 EAGAIN 错误）
 *  在for循环比较重要的一点是在读取队列中的所有消息之前重新注册接收消息通知。如果颠倒了顺序，如按照下面的顺序：队列中的所有消息都被读取了，while 循环终止；另一个消息被添加到了队列中；mq\_notify()被调用以重新注册接收消息通知。此刻，系统将不会产生新的通知信号，因为队列已经非空了，其结果是程序在下次调用 sigsuspend()时会永远阻塞

## 通过线程接收通知 ##

//mq_notify_thread.c
    // mq_notify_sig.c
    #include <stdio.h>
    #include <string.h>
    #include <stdlib.h>
    #include <mqueue.h>
    #include <sys/stat.h>
    #include <fcntl.h>
    #include <getopt.h>
    #include <zconf.h>
    #include <signal.h>
    #include <mqueue.h>
    #include <fcntl.h>
    #include <errno.h>
    #include <pthread.h>
    #include <mqueue.h>
    #include <signal.h>
    #include <fcntl.h>
    
    
    static void notifySetup(mqd_t *mqdp);
    static void threadFunc(union sigval sv)      /* Thread notification function */
    {
        
        ssize_t numRead;
        mqd_t *mqdp;
        void *buffer;
        struct mq_attr attr;
    
        mqdp = (mqd_t *)(sv.sival_ptr);
    
        /* Determine mq_msgsize for message queue, and allocate an input buffer
           of that size */
    
        if (mq_getattr(*mqdp, &attr) == -1){
        
            perror("mq_getattr");
            exit(EXIT_FAILURE);
        }
    
    
        buffer = malloc(attr.mq_msgsize);
        if (buffer == NULL){
        
            perror("malloc");
            exit(EXIT_FAILURE);
        }
    
    
        /* Reregister for message notification */
        // 当消息通知发生时，程序会在清空队列之前重新启用通知
        notifySetup(mqdp);
    
        while ((numRead = mq_receive(*mqdp, (char *)buffer, attr.mq_msgsize, NULL)) >= 0)
            printf("Read %ld bytes\n", (long) numRead);
        /*FIXME: above: should use %zd here, and remove (long) cast */
    
        if (errno != EAGAIN)                        /* Unexpected error */
        {
        
            printf("mq_receive");
            exit(EXIT_FAILURE);
        }
    
        free(buffer);
    }
    static void notifySetup(mqd_t *mqdp)
    {
        
        struct sigevent sev;
    
        sev.sigev_notify = SIGEV_THREAD;            //程序通过一个线程来请求通知需要将传入 mq_notify()的 sigevent 结构的 sigev_notify字段的值指定为 SIGEV_THREAD 。
        sev.sigev_notify_function = threadFunc;    //线程的启动函数 threadFunc() 是通过sigev_notify_function 字段来指定的
        sev.sigev_notify_attributes = NULL;
        /* Could be pointer to pthread_attr_t structure */
        sev.sigev_value.sival_ptr = mqdp;           /* Argument to threadFunc() */
    
        if (mq_notify(*mqdp, &sev) == -1){
        
            printf("mq_notify");
            exit(EXIT_FAILURE);
        }
    
    }
    int main(int argc, char *argv[])
    {
        
        mqd_t mqd;
    
        if (argc != 2 || strcmp(argv[1], "--help") == 0){
        
            printf("%s mq-name\n", argv[0]);
            exit(EXIT_FAILURE);
        }
    
        // 采用了非阻塞模式使得在接收到一个通知之后可以在无需阻塞的情况下完全清空队列
        mqd = mq_open(argv[1], O_RDONLY | O_NONBLOCK);
        if (mqd == (mqd_t) -1){
        
            printf("mq_open");
            exit(EXIT_FAILURE);
        }
    
    
        notifySetup(&mqd);
        pause();                    /* Wait for notifications via thread function */
        // 在启用消息通知之后，主程序会永远中止；定时器通知是通过在一个单独的线程中调用 threadFunc()来分发的①。
    }

# Linux 特有的特性 #

POSIX 消息队列在 Linux 上的实现提供了一些非标准的却相当有用的特性

## 通过命令行显示和删除消息队列对象 ##

POSIX IPC对象被实现成了虚拟文件系统中的文件，并且可以使用ls和rm来列出和删除这些文件。为列出和删除POSIX消息队列就必须要使用乤命令来将消息队列挂载到文件系统中：

$ mount  -t mqueue source target

source 可以是任意一个名字（通常将其指定为字符串 none），其唯一的意义是它将出现在/proc/mounts 中并且 mount 和 df 命令会显示出这个名字。target 是消息队列文件系统的挂载点。

下面的 shell 会话显示了如何挂载消息队列文件系统和显示其内容。首先为文件系统创建一个挂载点并挂载它  
![在这里插入图片描述][20210606213346709.png]  
接着显示新挂载在/proc/mounts 中的记录，然后显示挂载目录上的权限  
![在这里插入图片描述][20210606213439712.png]  
在 ls 命令的输出中需要注意的一点是消息队列文件系统在挂载时会自动为挂载目录设置粘滞位。（从 ls 的输出中的 other-execute 权限字段中有一个 t 就可以看出这一点。）这意味着非特权进程只能在它所拥有的消息队列上执行断开链接的操作。

接着创建一个消息队列，使用 ls 来表明它在文件系统中是可见的，然后删除该消息队列  
![在这里插入图片描述][20210606213642197.png]

## 获取消息队列的相关信息 ##

可以显示消息队列文件系统中的文件的内容，每个虚拟文件都包含了其关联的消息队列的相关信息：  
![在这里插入图片描述][20210606213819712.png]  
QSIZE字段的值为队列中所有数据的总字节数，剩下的字段则与消息通知相关。如果NOTIFY\_PID为非0，那么进程ID为该值的进程已经向该队列注册接收消息通知了，剩下的字段则提供了与这种通知相关的信息。

*  NOTIFY 是一个与其中一个 sigev\_notify 常量对应的值：0 表示 SIGEV\_SIGNAL，1表示 SIGEV\_NONE，2 表示 SIGEV\_THREAD。
 *  如果通知方式是 SIGEV\_SIGNAL，那么 SIGNO 字段指出了哪个信号会用来分发消息通知  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXpoZW5nZ3Vhbg_size_16_color_FFFFFF_t_70 2]

## 使用另一种 I/O 模型操作消息队列 ##

在Linux中，消息队列描述符实际上是一个文件描述符，因此可以使用IO多路复用系统调用（select()和 poll()）或 epoll API 来监控这个文件描述符

# 消息队列限制 #

SUSv3 为 POSIX 消息队列定义了两个限制

*  MQ\_PRIO\_MAX：定义了一条消息的最大优先级
 *  MQ\_OPEN\_MAX：
    
     *  一个实现可以定义这个限制来指明一个进程最多能打开的消息队列数量。
     *  SUSv3 要求这个限制最小为\_POSIX\_MQ\_OPEN\_MAX（8）。
     *  Linux 并没有定义这个限制，相反，由于 Linux将消息队列描述符实现成了文件描述符，因此适用于文件描述符的限制将适用于消息队列描述符。（换句话说，在 Linux 上，每个进程以及系统所能打开的文件描述符的数量限制实际上会应用于文件描述符数量和消息队列描述符数量之和。）

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXpoZW5nZ3Vhbg_size_16_color_FFFFFF_t_70]: /images/20221005/89da0dfbd913422b99f361f90b1fca27.png
[20210606201618800.png]: /images/20221005/83b7f6182fc04b36844c73dbb8957e2e.png
[20210606203935478.png]: /images/20221005/02bc33abc5404195bd3e6ed9f0a2f04f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXpoZW5nZ3Vhbg_size_16_color_FFFFFF_t_70 1]: /images/20221005/ae7b664cb5c8456cb45aa7588f81df0a.png
[20210606204040504.png]: /images/20221005/8646d810edbf4aceb94663b558248bf5.png
[2021060620405587.png]: /images/20221005/5c25ca6995c44664be16c61696cfa969.png
[20210606213346709.png]: /images/20221005/d6f21450e46246cf89e2c57ca7e89516.png
[20210606213439712.png]: /images/20221005/c6779b156e4f4045b48ed1f769f567ea.png
[20210606213642197.png]: /images/20221005/bac3c452f5bc4984ae32c3b373188612.png
[20210606213819712.png]: /images/20221005/6c9795f7d09c464e8d5547d2c685c00f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXpoZW5nZ3Vhbg_size_16_color_FFFFFF_t_70 2]: /images/20221005/9def421165f545f099fe0d2991a7285d.png