IPC 是 Linux 编程中一个重要的概念，IPC 有多种方式，本文主要介绍消息队列(Message Queues)，消息队列可以完成同一台计算机上的进程之间的通信，相比较管道，消息队列要复杂一些，但使用起来更加灵活和方便，Linux 既支持 UNIX SYSTEM V 的消息队列，也支持 POSIX 的消息队列，本文针对 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> 中：

    1
2
3

    #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 命令查看；

    Screenshot of ipc-l command

  • 上述三个对消息队列做出限制的当前值，也可以在 proc 文件系统中找到：

    Screenshot of proc file

  • 使用 sysctl 命令也可以查看消息队列的限制值：

    Screenshot of sysctl command

  • 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 的函数调用： 
      1

      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

    1
2
3

    #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

    1
2
3
4

    #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 的示例代码：

     1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

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

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

  • SYSTEM V 的 IPC 并不涉及文件系统，所以其设置的读写权限是不会受到 umask 的影响的，这点和 POSIX 的 IPC 有所不同。

3 向消息队列中发送消息

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

    1
2
3
4

    #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 指向要发送的消息，其消息结构的具体格式如下：

    1
2
3
4

    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 命令查到；

  • 运行截图，从截图中可以看出，消息队列中有三条消息：

    Screenshot of msg-send

4 从消息队列中读取消息

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

    1
2
3
4

    #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，这里再重复一遍：

    1
2
3
4

    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

  • 运行截图：

    Screenshot of msg-recv

5 消息队列的控制和操作

  • 函数：msgctl() - 对消息队列进行控制操作 
     1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

    struct ipc_perm {
    uid_t          cuid;   /* creator user ID */
    gid_t          cgid;   /* creator group ID */
    uid_t          uid;    /* owner user ID */
    gid_t          gid;    /* owner group ID */
    unsigned short mode;   /* r/w permissions */
};

struct msqid_ds {
    struct ipc_perm msg_perm;
    msgqnum_t       msg_qnum;    /* no of messages on queue */
    msglen_t        msg_qbytes;  /* bytes max on a queue */
    pid_t           msg_lspid;   /* PID of last msgsnd(2) call */
    pid_t           msg_lrpid;   /* PID of last msgrcv(2) call */
    time_t          msg_stime;   /* last msgsnd(2) time */
    time_t          msg_rtime;   /* last msgrcv(2) time */
    time_t          msg_ctime;   /* last change time */
};


#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 的消息队列 
    1

    msgctl(msqid, IPC_RMID, 0);
  • 下面的代码段将 ID 为 msqid 的消息队列的读写权限改为 0660 ： 
    1
2
3
4
5

    struct msqid_ds queue_ds;

msgctl(msqid, IPC_STAT, &queue_ds);
queue_ds.msg_perm.mode = 0660;
msgctl(msqid, IPC_SET, &queue_ds);
  • 在介绍 msgctl() 时介绍了 struct msqid_ds，
    • 这个结构是在建立一个消息队列时建立的，之后，内核将依靠这个结构维护这个消息队列；
    • 每个消息队列都对应着一个 struct msqid_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

  • 运行截屏：

    Screenshot of msg-recv

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> - 创建一个新的消息队列，其读写权限为指定的 <perm>。

8 其它

  • 前面介绍过，消息队列的实现在内核中是使用的链表，理论上，链表并不是一个先进先出(FIFO)的结构，但是消息队列的链表并没有向用户空间开放，同时内核也没有给出完全操作消息队列链表的接口，从操作上看，消息队列仍然具有先进先出的特性；

  • 内核并没有提供一个列出当前系统中所有消息队列的接口，只能依靠 proc 文件系统中的 /proc/sysvipc/msg 文件来查看当前所有消息队列的基本属性和权限，可以使用 cat /proc/sysvipc/msg 来查看消息队列的状态：

    Screenshot of proc filesystem

  • 源程序：listmsq.c(点击文件名下载源程序)演示了如何通过读取 /proc/sysvipc/msg 文件列出系统中所有的消息队列；
  • 还有一个办法可以列出所有的消息队列；因为消息队列的 ID 号的总量是有限制的(MSGMNI)，所有可以采用遍历所有消息队列的 ID 号的方法找到系统中所有的消息队列；
  • 源程序：traversemsq.c(点击文件名下载源程序)演示了如何通过遍历所有可能的消息队列 ID 号找到系统中所有的消息队列；
  • 尽管在本文的所有实例中，从消息队列传输的数据都是字符串，但消息队列是可以传输二进制数据的，也就是说，可以把一个结构完整第通过消息队列进行传输；
  • 管道与消息队列在进行数据交换上是不同的，消息队列是以消息为一个基本单位，而管道是一个字节流，举个例子说明其不同：

    假定进程 A 需要调用 func() 100 次，每调用 func() 一次，func() 都会将将 1 字节的运行结果发送给进程 B，所以，进程 B 一共会收到 100 字节的运行结果信息，如果使用消息队列来传递运行结果，func() 会向消息队列发送 100 个消息，而进程 B 也会接收 100 次消息；但是如果使用管道，func() 每次将 1 字节的运行结果送如管道，进程 B 可以不用马上接收，直到 func() 被调用 100 次后，进程 B 可以一次性地从管道中读出 100 个字节的运行结果；在这种情形下，显然，管道的效率更高一些。

  • 有关进程间通信(IPC)的的其它文章：

欢迎订阅 『进程间通信专栏』

欢迎访问我的博客：https://whowin.cn

email: hengch@163.com

donation