APUE读书笔记-15进程内部通信-07消息队列

++++++APUE读书笔记-15进程内部通信-07消息队列++++++

 

7、消息队列
================================================
 消息队列是存放在内核中的消息的链表，通过消息队列标识进行标记。我们把消息队列称为队列，把它的标识称为队列标识。
 另外Single UNIX Specification有一个实时扩展的消息队列实现，这里对实时扩展不进行讨论。
 通过msgget一个新的队列会被创建或者一个已经存在的队列会被打开。新的消息通过msgsnd函数被添加到队列的尾部。每个消息有一个正的长整数类型字段，一个非负的长度，以及实际的数据字节（和长度相关的）。所有这些在消息被添加到队列中的时候通过msgsnd来指定。通过msgrcv可以从一个队列中接收消息。我们没有必要以先进先出的次序来获取消息。其实，我们可以基于消息的类型字段来获取消息。
 每个队列有如下的msqid_ds结构和它相关联：
 struct msqid_ds {
  struct ipc_perm  msg_perm;     /* see Section 15.6.2 */
  msgqnum_t        msg_qnum;     /* # of messages on queue */
  msglen_t         msg_qbytes;   /* max # of bytes on queue */
  pid_t            msg_lspid;    /* pid of last msgsnd() */
  pid_t            msg_lrpid;    /* pid of last msgrcv() */
  time_t           msg_stime;    /* last-msgsnd() time */
  time_t           msg_rtime;    /* last-msgrcv() time */
  time_t           msg_ctime;    /* last-change time */
  .
 };
 这个结构定义了当前的队列状态。其中的成员由Single UNIX Specification来定义，实现中还可以包含一些额外的没有在这个标准中定义的成员域。
 文中列出了一个表格用来展示影响队列的各种系统限制。这里就不给出了。有一点需要注意的是，有些限制可能会依赖其它的限制，例如Linux中消息队列中最大的队列数目和所有队列能够容纳的最大数据量会影响最大的消息量。（注意数据是被包含在消息量中的）另外，在Linux中，即使消息包含0个字节的数据，那么也会认为这个消息包含了一个字节的数据，这样来限制排入队列中的消息的数目。

 一般来说第一个函数应该是msgget，用来创建一个队列或者打开一个已经存在的队列。
 #include
 int msgget(key_t key, int flag);
 返回值：如果成功返回队列ID，如果错误返回1。
 在前面，我们描述了将关键字转化为标识的一些规则，也讨论了创建一个队列或者引用一个已经存在的队列。当一个新的队列被创建的时候，msqid_ds结构的如下成员会被初始化。
 a.ipc_perm结构如前面说的那样被初始化。这个结构的mode成员会被设置为相应的权限标志。这些权限标志的值前面也提到过。
 b.msg_qnum,msg_lspid, msg_lrpid, msg_stime,和msg_rtime都会被设置为0。
 c.msg_ctime被设置成当前的时间。
 d.msg_qbytes 被设置成系统的限制。
 成功的时候，msgget返回非负的队列ID，这个被接下来的三个消息队列函数来使用。

 msgctl对队列进行各种操作，这个函数以及后面相应的信号量和共享内存的函数(semctl和shmctl)就像ioctl函数（它是通过不同参数，进行各种操作的函数，一般在编写驱动程序的时候用这个函数以及其参数来实现相应硬件的特定的功能）那样(所谓可以容纳各种杂乱操作的"垃圾"函数)。
 #include
 int msgctl(int msqid, int cmd, struct msqid_ds *buf );
 返回：如果成功返回0，如果错误返回1。
 参数cmd用来指定对msqid相应的队列进行的操作,如下。
 IPC_STAT:表示获取队列的msqid_ds结构，并把它存放在buf所值的位置。
 IPC_SET:从buf所指的结构变量中拷贝如下的成员到队列相关的msqid_ds结构：msg_perm.uid, msg_perm.gid, msg_perm.mode, 和 msg_qbytes。这个命令只有当进程的有效用户ID等于msg_perm.cuid或者msg_perm.uid或者一个进程具有超级用户权限的时候才能够被执行。只有超级用户可以增加msg_qbytes的值。
 IPC_RMID:将消息队列中仍然存在的数据，以及消息队列本身从系统中移走。这个删除的操作是立即生效的。任何其他的进程如果仍然在使用这个消息队列，那么它将在下次尝试操作这个队列的时候获得一个EIDRM的错误。这个命令只有当进程的有效用户ID等于msg_perm.cuid或者msg_perm.uid或者一个进程具有超级用户权限的时候才能够被执行。
 后面我们将会看到，这三个命令(IPC_STAT, IPC_SET, 和 IPC_RMID)在信号量以及共享内存中也被提供了。

 数据通过调用msgsnd函数被放在消息队列中。
 #include
 int msgsnd(int msqid, const void *ptr, size_t nbytes, int flag);
 返回值：如果成功返回0，如果错误返回1。
 如我们前面所提到的，每个消息由一个正的长整数，一个非负的长度(nbytes)，以及一个实际的数据(和前面nbytes长度相应)。消息会被放置在队列的尾部。
 ptr参数指向一个长整数，这个整数包含消息类型（正数），以及接下来紧跟着消息的数据。（如果nbytes为0那么没有消息数据）如果我们发送的最大的消息是512字节，我们可以定义如下的结构：
 struct mymesg {
  long  mtype;      /* 消息类型，正数 */
  char  mtext[512]; /* 消息数据，长度为nbytes */
 };
 然后让ptr参数指向mymesg结构。消息类型会在接收者以非先进先出方式接收数据的时候被使用。
 有些系统同时支持64位环境。这会影响长整数和指针类型的大小。例如：在64位的SPARC系统中，Solaris允许32位和64位的应用程序共存。如果一个32位的应用程序通过管道或者套接字和一个64位的应用程序交换这个结构的数据，那么就会出现问题。也就是说，一个32位应用程序可能会期望mtext成员从这个结构的第4个字节开始，而64位的应用程序会期望mtext成员从这个结构的第8个字节开始。这个时候，64位应用程序中的mtype成员的部分内容(即后4个字节)将会被32位应用程序当做mtext成员的部分内容，而32位应用程序的mtext成员的部分内容(即前4个字节)被64位应用程序当做mtype成员的部分内容。
 然而，这个问题在XSI的消息队列中不会发生。Solaris将32位版本的IPC系统调用用与64位版本的IPC系统调用不同的入口(应该是系统调用的中断入口)来进行实现。系统调用知道如何处理32位应用程序和64位应用程序之间的通信，并且特殊对待type成员防止与消息的数据部分相互干扰。唯一的潜在问题就是可能64位应用程序通过8字节的类型成员发送消息，而类型成员的值超过了32位应用程序能够用4字节表示的范围。这个时候，32位的应用程序将会看到一个被截断的类型成员。
 可以给flag指定IPC_NOWAIT值。这个效果类似文件I/O时候指定非阻塞标志。如果消息队列已经满了（或者是队列中消息的总数达到了系统的限制，或者是数据量的字节达到了系统的限制），那么指定IPC_NOWAIT将会导致msgsnd立即返回，错误为EAGAIN。如果IPC_NOWAIT没有被指定，那么我们会阻塞，一直到有空间可以容纳这个消息，或者消息队列从系统中被删除，或者捕获到一个信号并且信号处理函数返回。在第2个情况下，会返回一个EIDRM错误；在最后一个情况下，会返回EINTR错误。
 需要注意当不计考虑地删除一个消息队列时候，系统是如何处理这个情况的。因为每个消息队列中没有相应的引用计数（而对于打开的文件中却有引用计数），删除一个队列会导致使用这个队列的进程当下次对这个队列进行操作的时候，产生错误。信号量处理删除的方式也是这样。相反，当一个文件被删除的时候，文件的内容并没有被删除，直到最后一个打开的文件描述符号被删除。
 当msgsnd成功返回的时候，和消息队列相关的msqid_ds结构会被更新，通过msg_lspid来指明调用这个函数的进程ID，以及通过msg_stime指明调用这个函数的时间，还有通过msg_qnum来指明多了一个消息。

 如果从一个队列中接收一个消息，那么使用msgrcv函数。
 #include
 ssize_t msgrcv(int msqid, void *ptr, size_t nbytes , long type, int flag);
 返回值：如果成功返回消息数据部分的大小，如果错误返回1。
 类似msgsnd，参数ptr指向一个长整类型(消息的类型字段存放在其中)后面接着一个存放实际数据的缓存。参数nbytes指定数据缓存的大小。如果返回的消息比nbytes大并且MSG_NOERROR已经被设置，那么消息会被截断（这个时候我们不会收到消息被截断的通知同时剩余的消息也会被忽略）。如果消息太大，并且这个MSG_NOERROR没有被设置，那么会返回E2BIG错误（同时消息会保留在队列中）。
 我们可以通过type参数指定我们想要什么类型的消息。
 type == 0:表示将会返回队列中的第一个消息。
 type > 0: 表示将会返回队列中消息类型的值等于type参数的第一个消息。
 type < 0:表示将会返回队列中第一个最小的消息的类型值，这个值小于或者等于type的绝对值。
 非0的type表示将会以非先进先出的次序读取消息。例如type可以表示一个优先权，应用程序可以给消息赋予相应的优先级。另外一个使用这个成员的地方就是让这个成员包含一个进程的ID，这在多个客户和一个服务进程使用单个消息队列进行通信的时候，会用到(只要进程ID适合长整数类型)。
 我们可以指定一个IPC_NOWAIT值给flag，这样可以使得这个操作成为非阻塞的操作，导致当队列中没有指定类型的消息的时候，msgrcv会立即返回-1并且设置errno为ENOMSG。如果没有指定IPC_NOWAIT，那么这个操作会一直阻塞，直到一个指定类型的消息可用，或者队列在系统中被删除（返回-1并且设置errno为EIDRM），或者捕获到一个信号并且信号处理函数返回（导致msgrcv返回1并且设置errno为EINTR）。
 当msgrcv成功的时候，内核会更新相应消息队列的msqid_ds结构变量，通过msg_lrpid表示最后调用这个函数的进程PID，通过msg_rtime表示调用的时间，通过msg_qnum表示队列中少了一个消息。

 消息队列和管道以及流的时间对比的例子
 我们可以通过消息队列或者全双工管道来实现客户和服务进程之间的双向通信。
 下面的表格给出了Solaris上面三种类型通信技术的时间对比：消息队列，基于流的管道，以及UNIX域套接字。测试程序创建一个IPC通道，调用fork然后由父进程向子进程发送200兆数据。通过调用100,000次msgsnd来发送数据，消息队列的消息长度为2000字节；还有通过100,000次调用write给基于流的管道和UNIX域的套接字，每次write的字节数目为2000字节。下面的时间以秒为单位。
  对比的表格数据如下：
+---------------------------------------------+
|      Operation      | User | System | Clock |
|---------------------+------+--------+-------|
| message queue       | 0.57 |  3.63  | 4.22  |
|---------------------+------+--------+-------|
| STREAMS pipe        | 0.50 |  3.21  | 3.71  |
|---------------------+------+--------+-------|
| UNIX domain socket  | 0.43 |  4.45  | 5.59  |
+---------------------------------------------+

 这些数据给我们展示的结果表示，本来想要提供更高速度的消息队列，其实它的速度没有其他形式的IPC的速度快（实际上基于流的管道的速度比消息队列还快）。（当实现消息队列之后，其他形式的IPC可用的只有半双工的管道???）当我们考虑到前面节提到的消息队列的优点和缺点问题的时候，我们得到这样一个结论：在新的应用程序中，我们不应当使用消息队列了。

参考：