学习在 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 <sys/ipc.h>

include <sys/msg.h>

include

include

include “queue.h”

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

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

int qid = msgget(key, 0666 | IPC_CREAT);
if (qid < 0) report_and_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 %in", 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 <sys/ipc.h>

include <sys/msg.h>

include

include “queue.h”

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

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

int qid = msgget(key, 0666 | IPC_CREAT); / access if created already /
if (qid < 0) report_and_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], MSG_NOERROR | IPC_NOWAIT) < 0)
puts(“msgrcv trouble…”);
printf(“%s received as type %in”, msg.payload, (int) msg.type);
}

/ remove the queue /
if (msgctl(qid, IPC_RMID, NULL) < 0) / NULL = ‘no flags’ /
report_and_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](https://opensource.com/users/mkalindepauledu) 选题：[lujun9972](https://github.com/lujun9972) 译者：[FSSlc](https://github.com/FSSlc) 校对：[wxy](https://github.com/wxy)

本文由 [LCTT](https://github.com/LCTT/TranslateProject) 原创编译，[Linux中国](https://linux.cn/) 荣誉推出