学习在 Linux 中进程是如何与其他进程进行同步的。

本篇是 Linux 下进程间通信（IPC）系列的第二篇文章。第一篇文章 聚焦于通过共享文件和共享内存段这样的共享存储来进行 IPC。这篇文件的重点将转向管道，它是连接需要通信的进程之间的通道。管道拥有一个写端用于写入字节数据，还有一个读端用于按照先入先出的顺序读入这些字节数据。而这些字节数据可能代表任何东西：数字、员工记录、数字电影等等。

管道有两种类型，命名管道和无名管道，都可以交互式的在命令行或程序中使用它们；相关的例子在下面展示。这篇文章也将介绍内存队列，尽管它们有些过时了，但它们不应该受这样的待遇。

在本系列的第一篇文章中的示例代码承认了在 IPC 中可能受到竞争条件（不管是基于文件的还是基于内存的）的威胁。自然地我们也会考虑基于管道的 IPC 的安全并发问题，这个也将在本文中提及。针对管道和内存队列的例子将会使用 POSIX 推荐使用的 API，POSIX 的一个核心目标就是线程安全。

请查看一些 mq_open 函数的 man 页，这个函数属于内存队列的 API。这个 man 页中有关 特性 的章节带有一个小表格：

| 接口 | 特性 | 值 |
| | >|3|>|2|>receiver
+-+ +-+ +-+ +-+
“`

在上面展示的 4 个消息中，标记为 1 的是开头，即最接近接收端，然后另个标记为 2 的消息，最后接着一个标记为 3 的消息。假如按照严格的 FIFO 行为执行，消息将会以 1-2-2-3 这样的次序被接收。但是消息队列允许其他收取次序。例如，消息可以被接收方以 3-2-1-2 的次序接收。

mqueue 示例包含两个程序，sender 将向消息队列中写入数据，而 receiver 将从这个队列中读取数据。这两个程序都包含的头文件 queue.h 如下所示：

示例 4. 头文件 queue.h

“`

define ProjectId 123

define PathName “queue.h” /* any existing, accessible file would do */

define MsgLen 4

define MsgCount 6

typedef struct {
long type; /* must be of type long /
char payload[MsgLen + 1]; / bytes in the message */
} queuedMessage;
“`

上面的头文件定义了一个名为 queuedMessage 的结构类型，它带有 payload（字节数组）和 type（整数）这两个域。该文件也定义了一些符号常数（使用 #define 语句），前两个常数被用来生成一个 key，而这个 key 反过来被用来获取一个消息队列的 ID。ProjectId 可以是任何正整数值，而 PathName 必须是一个存在的、可访问的文件，在这个示例中，指的是文件 queue.h。在 sender 和 receiver 中，它们都有的设定语句为：


key_t key = ftok(PathName, ProjectId); /* generate key */
int qid = msgget(key, 0666 | IPC_CREAT); /* use key to get queue id */


ID qid 在效果上是消息队列文件描述符的对应物。

示例 5. sender 程序

“`

include

include

include

include

include

include “queue.h”

void reportandexit(const char* msg) {
perror(msg);
exit(-1); /* EXIT_FAILURE */
}

int main() {
keyt key = ftok(PathName, ProjectId);
if (key < 0) reportand_exit(“couldn’t get key…”);

int qid = msgget(key, 0666 | IPCCREAT);
if (qid < 0) reportand_exit(“couldn’t get queue id…”);

char* payloads[] = {“msg1”, “msg2”, “msg3”, “msg4”, “msg5”, “msg6”};
int types[] = {1, 1, 2, 2, 3, 3}; /* each must be > 0 /
int i;
for (i = 0; i < MsgCount; i++) {
/ build the message */
queuedMessage msg;
msg.type = types[i];
strcpy(msg.payload, payloads[i]);

/* send the message */
msgsnd(qid, &msg, sizeof(msg), IPC_NOWAIT); /* don't block */
printf("%s sent as type %i\n", msg.payload, (int) msg.type);

}
return 0;
}
“`

上面的 sender 程序将发送出 6 个消息，每两个为一个类型：前两个是类型 1，接着的连个是类型 2，最后的两个为类型 3。发送的语句：


msgsnd(qid, &msg, sizeof(msg), IPC_NOWAIT);


被配置为非阻塞的（IPC_NOWAIT 标志），是因为这里的消息体量上都很小。唯一的危险在于一个完整的序列将可能导致发送失败，而这个例子不会。下面的 receiver 程序也将使用 IPC_NOWAIT 标志来接收消息。

示例 6. receiver 程序

“`

include

include

include

include

include “queue.h”

void reportandexit(const char* msg) {
perror(msg);
exit(-1); /* EXIT_FAILURE */
}

int main() {
keyt key= ftok(PathName, ProjectId); /* key to identify the queue */
if (key < 0) reportand_exit(“key not gotten…”);

int qid = msgget(key, 0666 | IPCCREAT); /* access if created already */
if (qid < 0) reportand_exit(“no access to queue…”);

int types[] = {3, 1, 2, 1, 3, 2}; /* different than in sender /
int i;
for (i = 0; i < MsgCount; i++) {
queuedMessage msg; / defined in queue.h */
if (msgrcv(qid, &msg, sizeof(msg), types[i], MSGNOERROR | IPCNOWAIT) < 0)
puts(“msgrcv trouble…”);
printf(“%s received as type %i\n”, msg.payload, (int) msg.type);
}

/* remove the queue */
if (msgctl(qid, IPCRMID, NULL) < 0) /* NULL = ‘no flags’ */
reportand_exit(“trouble removing queue…”);

return 0;
}
“`

这个 receiver 程序不会创建消息队列，尽管 API 尽管建议那样。在 receiver 中，对


int qid = msgget(key, 0666 | IPC_CREAT);


的调用可能因为带有 IPC_CREAT 标志而具有误导性，但是这个标志的真实意义是如果需要就创建，否则直接获取。sender 程序调用 msgsnd 来发送消息，而 receiver 调用 msgrcv 来接收它们。在这个例子中，sender 以 1-1-2-2-3-3 的次序发送消息，但 receiver 接收它们的次序为 3-1-2-1-3-2，这显示消息队列没有被严格的 FIFO 行为所拘泥：

“`
% ./sender
msg1 sent as type 1
msg2 sent as type 1
msg3 sent as type 2
msg4 sent as type 2
msg5 sent as type 3
msg6 sent as type 3

% ./receiver
msg5 received as type 3
msg1 received as type 1
msg3 received as type 2
msg2 received as type 1
msg6 received as type 3
msg4 received as type 2
“`

上面的输出显示 sender 和 receiver 可以在同一个终端中启动。输出也显示消息队列是持久的，即便 sender 进程在完成创建队列、向队列写数据、然后退出的整个过程后，该队列仍然存在。只有在 receiver 进程显式地调用 msgctl 来移除该队列，这个队列才会消失：


if (msgctl(qid, IPC_RMID, NULL) < 0) /* remove queue */


总结

管道和消息队列的 API 在根本上来说都是单向的：一个进程写，然后另一个进程读。当然还存在双向命名管道的实现，但我认为这个 IPC 机制在它最为简单的时候反而是最佳的。正如前面提到的那样，消息队列已经不大受欢迎了，尽管没有找到什么特别好的原因来解释这个现象；而队列仍然是 IPC 工具箱中的一个工具。这个快速的 IPC 工具箱之旅将以第 3 部分（通过套接字和信号来示例 IPC）来终结。

via: https://opensource.com/article/19/4/interprocess-communication-linux-channels

作者：Marty Kalin 选题：lujun9972 译者：FSSlc 校对：wxy

本文由 LCTT 原创编译，Linux中国 荣誉推出