Linux 下的进程间通信:使用管道和消息队列

Marty Kalin 的头像

·

·

·

9,051 次阅读

学习在 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/) 荣誉推出

发表回复

您的电子邮箱地址不会被公开。 必填项已用 * 标注