IPC之三：使用 System V 消息队列进行进程间通信的实例

IPC 是 Linux 编程中一个重要的概念，IPC 有多种方式，本文主要介绍消息队列(Message Queues)，消息队列可以完成同一台计算机上的进程之间的通信，相比较管道，消息队列要复杂一些，但使用起来更加灵活和方便，本文针对 System V 消息队列，并给出了多个具体的实例，每个实例均附有完整的源代码；本文所有实例在 Ubuntu 20.04 上编译测试通过，gcc版本号为：9.4.0；本文适合 Linux 编程的初学者阅读。

1 消息队列的基本概念

消息队列是 Linux 下多种 IPC 方法之一，最早源自 UNIX System V，后来 POSIX 也制定了消息队列的标准；
所以，实际上有两种不同类型的消息队列：System V 消息队列和 POSIX 消息队列，两种消息队列的功能几乎是一样的，但调用方法不同；
Linux 既支持 Sytem V 消息队列也支持 POSIX 消息队列，本文针对 System V 消息队列；
消息队列本质上是存放在内核空间中的一个消息链表，由内核负责维护，这个链表显然是要占用内核资源的，所以对消息队列是有一定限制的，以避免过多地占用内核资源，这些限制定义在头文件 <linux/msg.h> 中：
```
#define MSGMNI 32000   /* <= IPCMNI */     /* max # of msg queue identifiers */
#define MSGMAX  8192   /* <= INT_MAX */   /* max size of message (bytes) */
#define MSGMNB 16384   /* <= INT_MAX */   /* default max size of a message queue */
```
- 头文件中对这三个宏定义有比较详细的说明；
- MSGMNI 是系统中消息队列数量的上限；
- MSGMAX 是单条消息的最大字节数；这个值设置为 MSGMNB 的 1/2，以保证在消息队列中至少可以有两条消息；
- MSGMNB 是每个消息队列所占用的最大字节数，也就是一个队列的所有消息的长度之和不能大于这个数；
上述三个对消息队列做出限制的当前值可以简单地使用 ipcs -q -l 命令查看；
上述三个对消息队列做出限制的当前值，也可以在 proc 文件系统中找到：
使用 sysctl 命令也可以查看消息队列的限制值：

IPC 标识符
- System V 定义了三种 IPC 方法：消息队列(Message queues)、信号量(Semaphores)和共享内存(Shared Memory)，这也是 Linux 下的三种重要的 IPC 方法，我们把这些统称为"IPC 对象(IPC Object)"；
- 每个 IPC 对象都有一个与之关联的唯一的标识符(IPC Identifier)，这个标识符在内核中用于唯一地标识一个 IPC 对象；例如，要访问特定的一个消息队列，只需要知道这个消息队列的标识符(ID)即可；
- 在 Ubuntu 上，这个 ID 的唯一性与 IPC 对象的类型是无关的，比如：一个消息队列的 ID 是 1234，那么，在共享内存中就不可能有 ID 为 1234 的 IPC 对象；
- 在下面的描述中 IPC 标识符 ID 将被简称为 ID；
- 消息队列的 ID，在 Linux 下其实就是一个 32 位整数的序列号，从 0 开始；也就是说，系统建立的第一个 IPC 对象的 ID 是 0，第二个 IPC 对象的 ID 是 1，…，以此类推；
- 即便一个 IPC 对象被删除，其对应的 ID 号空闲出来，新建立的 IPC 对象也不会用这个空闲出来的 ID 号，而是按照序列号继续延续，也就是说，系统建立的第一个 IPC 对象的 ID 是 0，…，第三个 IPC 对象的 ID 是 2，现在删除 ID 号为 1 的 IPC 对象，再建立一个 IPC 对象时，其 ID 号是 3 而不会是 1。
IPC Key
- Unix 有一个所谓“一切皆文件”的原则，但是 System V IPC 的设计上似乎与这一原则有些偏差，内核使用 “IPC 标识符 ID” 来标识一个 IPC 对象，但是我们并不能通过打开一个文件来获得与这个 “IPC 标识符 ID” 的关联；
- IPC 需要使用一个“IPC Key”来创建或者获取 “IPC 标识符 ID”，理论上说，这个 IPC Key 可以使用任意方法生成；
- 同时 System V IPC 提供了一个可以生成 IPC Key 的函数调用：
```
key_t ftok(const char *pathname, int proj_id);
```
- 该调用实际上是将 pathname 在内核中的 inode 节点号、设备号以及 proj_id 三者组合到一起生成 IPC Key，这使得 IPC Key 和一个文件路径产生了联系，似乎又有点回到“一切皆文件”的原则上来；
- 总之，System V IPC 的逻辑是：
  - 用 路径 + 项目 ID(一个整数) 生成一个 IPC Key；
  - 通过 Key 生成或者获取到 IPC 标识符 ID；
  - 用 IPC 标识符 ID 对 IPC 对象进行操作；
- 举例：key_t key = ftok("./tmp/", 1234);

2 创建/获取消息队列 ID

消息队列 ID 是操作一个消息队列的唯一标识；
创建一个新的消息队列会为这个消息队列生成一个 ID，要操作一个已经存在的消息队列也必须首先获得该消息队列的 ID；
下面两个函数用于创建/获取消息队列 ID；
函数：ftok() - 将路径名和项目 ID 组合转换为 IPC Key，此值将用于创建/获取一个消息队列 ID
```
#include <sys/ipc.h>key_t ftok(const char *pathname, int proj_id);
```
- 这个函数在前面已经提到过，这里稍微详细地介绍一下；
- pathname 既可以是一个目录的路径，比如：/tmp/，也可以是一个文件的路径，比如：/tmp/msgqueue.txt，既可以是一个绝对路径，也可以是一个相对路径；
- proj_id 称为项目 ID，实际上就是个数字，在这个函数中，仅使用这个数字的低 8 位；
- 通常认为，同一个项目的程序总是放在同一个目录下，这个目录可以指定为 pathname，但同一个目录下可能有不同的几个项目，那么可以用项目 ID 来区分，这样就可以生成不同的 Key，从而创建/获取不同的消息队列的 ID；
- 该函数会取出 pathname 的 inode 节点号和设备号(dev_no)，然后与 proj_id 组合生成一个 key，同样的 pathname 和 proj_id 将获得相同的 key；
- 生成 key 的组合方式为：(inode & 0xffff) | ((dev_no & 0xff) << 16) | ((proj_id & 0xff) << 24)
- 源程序：ipc-key.c(点击文件名下载源程序)验证了这种组合方式的正确性。
函数：msgget() - 创建新消息队列，或者获取已有消息队列的 ID
```
#include <sys/ipc.h>
#include <sys/msg.h>int msgget(key_t key, int msgflg);
```
- 该系统调用用于建立一个新的消息队列，或者获取一个已经存在的消息队列的 ID；
- 该系统调用返回的 ID 号将成为对一个消息队列操作的唯一标识；
- 该系统调用返回与 key 参数值相关联的消息队列标识符 ID；它既可以用于获取已经存在的消息队列的标识符，也可以用于创建一个新的消息队列；
- msgget() 中的 key 通常使用 ftok() 生成(也可以自定义生成)，也可以设置为 IPC_PRIVATE；
- msgflag 的有效值有 IPC_CREAT 和 IPC_EXCL：
  - 当 IPC_CREAT 时，如果 key 对应的消息队列存在，则返回其消息队列的 ID，如果 key 对应的消息队列不存在，则建立与 key 关联的消息队列，并返回消息队列的 ID；
  - 当 IPC_CREAT | IPC_EXCL 时，如果 key 对应的消息队列存在，则报错返回 -1，errno = EEXIST(File exists)；如果 key 对应的消息队列不存在，则建立与 key 关联的消息队列，并返回消息队列的 ID；
  - 当 IPC_EXEL 时，如果 key 对应的消息队列存在，则返回消息队列的 ID(这点和 IPC_CREAT 一样)，如果 key 对应的消息队列不存在，则返回 -1，errno = ENOENT(No such file or directory)；
  - 另外，msgflag 还可以加上所创建的消息队列的读写权限，比如：0666；
  - msgflag 举例：IPC_CREAT | IPC_EXEL | 0666
- 当 key = IPC_PRIVATE 时，msgget() 将创建一个新的消息队列并返回该消息队列的 ID；
  - 这样生成的消息队列只有 ID，没有 key(key 为 0)，所以其它进程并不能方便地使用这个消息队列，通常只能在子进程之间使用；
  - 实际上，IPC_PRIVATE 的值是 0，所以我们自己生成的 key 不能是 0，否则相当于将 key 设置为 IPC_PRIVATE；
- 再次强调消息队列的生成逻辑：通过“一个路径 + 项目 ID”生成 key，通过 key 与一个消息队列 ID 相对应，通过消息队列 ID 操作消息队列。

创建/获取消息队列 ID 的示例代码：

#include <sys/ipc.h>
#include <sys/msg.h>
...
key_t key = ftok("/tmp/", 1234);
// check if the key exists
msqid = msgget(ipc_key, IPC_EXCL);
if (msqid == -1) {// key doesn't exist, create itmsqid = msgget(ipc_key, IPC_CREAT | 0666);if (msqid == -1) {perror("msgget()");exit(EXIT_FAILURE);}
}
...

特别要注意的是 msgflag 参数中的读/写权限的设置，如果不显式标明，那么默认的读写权限将变成 0000，这样的一个消息队列是没有办法进行读写的；当然还可以使用 msgctl() 修改消息队列的读写权限，但需要更高的权限(比如root)才能做到。

3 向消息队列中发送消息

函数：msgsnd() - 向消息队列中发送一条消息

#include <sys/ipc.h>
#include <sys/msg.h>int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);

通过调用 getget() 获取了消息队列的 ID 后，就可以向这个消息队列中发送消息了；
该系统调用向 ID 为 msgid 的消息队列中发送一条消息，调用该函数的进程必须要有该消息队列的写权限；
该函数在调用成功时返回 0，调用失败时返回 -1，errno 中为错误代码；
msgp 指向要发送的消息，其消息结构的具体格式如下：
```
struct msgbuf {long mtype;       /* message type, must be > 0 */char mtext[1];    /* message data */
};
```
- mtype 字段必须是正整数值，接收进程可以根据这个字段对收到的消息进行选择；
- mtext 字段是一个数组(或其他结构)，其大小由 msgsnd() 调用中的 msgsz 指定；
msgsz 不能是负数，但可以为 0，也就是所谓零长度的消息(即没有 mtext 字段)；
msgflag 可以填 0 或者 IPC_NOWAIT
- msgsnd() 默认是阻塞的，当用于消息队列的内存空间不够用时，msgsnd() 会阻塞，直到有足够的空间可用；
- 当 msgflag 设置为 IPC_NOWAIT 时，msgsnd() 在消息队列的空间不够用时也不会阻塞，会立即返回失败，errno = EAGAIN；
还有以下两种情况会导致 msgsnd() 调用失败
1. msgid 指向的消息队列不存在或者已经被删除，返回失败，errno = EIDRM；
2. 调用该函数时被信号处理程序中断，此时，msgsnd() 不会自动重启，返回失败，errno = EINTR；
源程序：msg-send.c(点击文件名下载源程序)演示了如何使用 msgsnd() 向消息队列中发送消息；
这个程序以当前目录的相对路径 “./” 建立一个 key，然后用这个 key 建立/获取一个消息队列的 ID，然后向这个消息队列中发送了三条消息；
编译：gcc -Wall msg-send.c -o msg-send
运行：./msg-send
使用命令 ipcs -q 可以查看消息队列情况；
使用命令 ipcrm -q [ID] 可以删除指定详细队列中的消息，其中 [ID] 可以通过 ipcs -q 命令查到；
运行截图，从截图中可以看出，消息队列中有三条消息：

4 从消息队列中读取消息

函数：msgrcv() - 从消息队列中接收一条消息

#include <sys/ipc.h>
#include <sys/msg.h>ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp, int msgflg);

该系统调用从 ID 为 msgid 的消息队列中接收一条消息，调用该函数的进程必须要有该消息队列的读权限；
该系统调用从消息队列中取出一条消息放到 msgp 参数指定的缓冲区内，同时会从消息队列中删除这条消息；
该系统调用在调用成功时，返回收到消息的长度，调用失败时，返回 -1，errno 中为错误代码；
与 msgsnd() 类似，msgp 指向接收消息的缓冲区，其结构的具体格式同 msgsnd() 中介绍的 struct msgbuf，这里再重复一遍：
```
struct msgbuf {long mtype;       /* message type, must be > 0 */char mtext[1];    /* message data */
};
```
msgsz 参数为 msgp 指向的结构中 mtext 字段的长度；
当收到消息的长度大于 msgsz 时：
- 如果 msgflag 中设置了 MSG_NOERROR，则该消息将被截断，被截断的消息部分将丢失，调用返回成功；
- 如果 msgflag 中没有设置 MSG_NOERROR，则调用失败，errno = E2BIG；
msgtyp 用于选择收到的消息类别；收到的信息格式符合 struct msgbuf，其结构中的 mtype 表示该消息类别，msgtype 可以选择接收哪种类别的消息：
- 当 msgtyp = 0 时，从消息队列中读取出第一条消息；
- 当 msgtyp > 0 时，从消息队列中读取出 mtype == msgtyp 的第一条消息；
- 当 msgtyp < 0 时，从消息队列中读取出 mtype <= |msgtyp| 且 mtype 为最小的第一条消息；
  
  举个例子：消息队列中有 4 条消息，第 1 条消息的 mtype = 3，第 2 条消息的 mtype = 1，第 3 条消息的 mtype = 4，第 4 条消息的 mtype = 2，当使用 msgrcv(msqid, (void *)msgp, msgsz, -3, ...) 从消息队列中读取消息时，msgtyp = -3，|msgtyp| = 3，那么，第 1、2、4 条消息都满足 mtype <= |msgtyp| 的条件，但是收到的消息为第 2 条，因为第 2 条的 mtype 最小。
msgflag 可以有下面的值组合而成：
- IPC_NOWAIT：如果消息队列中没有请求类型(msgtyp)的消息，则立即返回。系统调用失败，errno = ENOMSG
- MSG_COPY：从消息队列中读取消息，但是不会破坏原消息队列，即：取出消息后，并不把这条消息从消息队列中删除；
  - 该标志必须和 IPC_NOWAIT 一起使用，当消息队列中没有符合条件的消息时，会立即返回调用失败，errno = ENOMSG
  - 该标志不能与 MSG_EXCEPT 一起使用；
- MSG_EXCEPT：当 msgtyp > 0 时，从消息队列中读取 mtype != msgtyp 的第一条消息；
- MSG_NOERROR：如果收到消息的长度大于参数 msgsz，则将消息截断；
如果消息队列中没有符合条件的消息，同时，msgflag 中没有设置 IPC_NOWAIT，则该调用将产生阻塞，直到：
- 一条所需类型(mtype)的消息进入到消息队列中；
- 该消息队列被删除，此时返回调用失败，errno = EIDRM
- 当前进程捕获了一个信号，此时返回调用失败，errno = EINTR
源程序：msg-recv.c(点击文件名下载源程序)演示了如何使用 msgrcv() 从消息队列中读取消息；
编译：gcc -Wall msg-recv.c -o msg-recv
与上一个示例 msg-send 一样，该程序以当前目录的相对路径 “./” 建立一个 key，然后用这个 key 获取一个消息队列的 ID，如果这个消息队列不存在，程序将退出；
运行：./msg-recv [msgtype]
- 当 [msgtype] 存在时，从消息队列中读取其消息类型为 [msgtype] 的第一条消息；
- 当 [msgtyoe] 不存在时，默认 [msgtyoe] 为 0；
这个程序可以和上一个示例 msg-send 一起使用，这时，msg-send 和 msg-recv 两个程序应该放在用一个工作目录下，并从这个目录下运行；
- msg-send 程序建立消息队列，并向这个队列中发送了三条消息，消息类型分别是 1、2 和 3；
- 运行 msg-recv 时，可以指定只读取其中一种类型的消息，比如：./msg-recv 2
运行截图：

5 消息队列的控制和操作

函数：msgctl() - 对消息队列进行控制操作

struct ipc_perm {__key_t __key;                  /* Key. */__uid_t uid;                    /* Owner's user ID. */__gid_t gid;                    /* Owner's group ID. */__uid_t cuid;                   /* Creator's user ID. */__gid_t cgid;                   /* Creator's group ID. */__mode_t mode;                  /* Read/write permission. */unsigned short int __seq;       /* Sequence number. */unsigned short int __pad2;__syscall_ulong_t __glibc_reserved1;__syscall_ulong_t __glibc_reserved2;
};struct msqid_ds {struct ipc_perm msg_perm;       /* structure describing operation permission */__MSQ_PAD_TIME (msg_stime, 1);  /* time of last msgsnd command */__MSQ_PAD_TIME (msg_rtime, 2);  /* time of last msgrcv command */__MSQ_PAD_TIME (msg_ctime, 3);  /* time of last change */__syscall_ulong_t __msg_cbytes; /* current number of bytes on queue */msgqnum_t msg_qnum;             /* number of messages currently on queue */msglen_t msg_qbytes;            /* max number of bytes allowed on queue */__pid_t msg_lspid;              /* pid of last msgsnd() */__pid_t msg_lrpid;              /* pid of last msgrcv() */__syscall_ulong_t __glibc_reserved4;__syscall_ulong_t __glibc_reserved5;
};#include <sys/ipc.h>
#include <sys/msg.h>int msgctl(int msqid, int cmd, struct msqid_ds *buf);

该系统调用对 msqid 指定的消息队列执行一个由 cmd 指定的命令，包括：获取/设置消息队列的属性，删除消息队列等；
Linux 下允许的 cmd 值如下：
- IPC_RMID - 删除一个消息队列
- IPC_SET - 设置消息队列的权限结构
- IPC_STAT - 获取消息队列的权限结构
当 cmd 为 IPC_STAT 时，该调用将消息队列在内核中的属性拷贝到 buf 指向的 struct msqid_ds 中；
当 cmd 为 IPC_SET 时，该调用将根据 buf 指向的 struct msqid_ds 的信息更新该消息队列在内核中的属性；
- 其实可以更新的属性有限，已知可更新的属性有：msg_qbytes、msg_perm.uid、msg_perm.gid 和 msg_perm.mode 的最后九位；
- 这个命令的主要用途就是改变一个消息队列的读写权限，也就是修改 msg_perm.mode 的最后九位；
下面代码可以删除 ID 为 msqid 的消息队列
```
msgctl(msqid, IPC_RMID, 0);
```

下面的代码段将 ID 为 msqid 的消息队列的读写权限改为 0660 ：

struct msqid_ds queue_ds;msgctl(msqid, IPC_STAT, &queue_ds);
queue_ds.msg_perm.mode = 0660;
msgctl(msqid, IPC_SET, &queue_ds);

6 实例：消息队列的小工具

最后这个实例可以作为消息队列的一个小工具使用，这个小工具有如下功能：
1. 向消息队列中发送一条新消息；
2. 从消息队列中读取一条消息；
3. 删除一个消息队列；
4. 修改消息队列的读写权限。
源程序：msgtool.c(点击文件名下载源程序)可以看作是一个消息队列的小实用工具；
- 以当前路径 + ‘m’ 的 ASCII 码组合成 key，该实用工具仅针对该 key 关联的消息队列进行操作；
- 在发送消息时如果不存在与 key 关联的消息队列则建立一个新的消息队列；
- 从消息队列中接收消息时，如果与 key 关联的消息队列不存在，则直接退出；
- 删除消息队列时，如果与 key 关联的消息队列不存在，则直接退出；
- 修改消息队列的读/写权限时，如果与 key 关联的消息队列不存在，则直接退出；
- 该程序使用了上面介绍过的与消息队列相关的所有函数。
编译：gcc -Wall msgtool.c -o msgtool
向消息队列中发送消息：./msgtool s 2 "Hello world!"
从消息队列中接收消息：./msgtool r
删除消息队列：./msgtool d
修改消息队列的读写权限：./msgtool m 0666
运行截屏：

7 操作消息队列的命令行命令

ipcs -q -l - 显示消息队列的限制值；
ipcs -q - 显示现有消息队列的 key、ID 等部分属性；
ipcs -q -i <ID> - 显示指定 ID 的消息队列的属性(比 ipcs -q 显示的属性要多些)；
ipcrm -Q <key> - 删除指定 key 的消息队列；
ipcrm -q <ID> - 删除指定 ID 的消息队列
ipcrm --all=msg - 删除所有的消息队列
ipcmk -Q - 创建一个新的消息队列，其读写权限为默认的 0644；
ipcmk -Q -p <perm> - 创建一个新的消息队列，其读写权限为指定的。

8 其它

前面介绍过，消息队列的实现在内核中是使用的链表，理论上，链表并不是一个先进先出(FIFO)的结构，但是消息队列的链表并没有向用户空间开放，同时内核也没有给出完全操作消息队列链表的接口，从操作上看，消息队列仍然具有先进先出的特性；
内核并没有提供一个列出当前系统中所有消息队列的接口，只能依靠 proc 文件系统中的 /proc/sysvipc/msg 文件来查看当前所有消息队列的基本属性和权限，可以使用 cat /proc/sysvipc/msg 来查看消息队列的状态：

源程序：listmsq.c(点击文件名下载源程序)演示了如何通过读取 /proc/sysvipc/msg 文件列出系统中所有的消息队列；
还有一个办法可以列出所有的消息队列；因为消息队列的 ID 号的总量是有限制的(MSGMNI)，所有可以采用遍历所有消息队列的 ID 号的方法找到系统中所有的消息队列；
源程序：traversemsq.c(点击文件名下载源程序)演示了如何通过遍历所有可能的消息队列 ID 号找到系统中所有的消息队列；