Linux 管道通信详解
一、进程间通信基础
进程间通信（IPC）的核心在于不同进程之间如何交换信息。常见需求包括数据传输、资源共享、事件通知以及进程控制。
在 Linux 系统中，主要的 IPC 方式包括管道、System V IPC 和 POSIX IPC。由于进程具有独立性，实现通信的前提是它们必须能访问同一份资源。
二、管道机制
1. 什么是管道
管道是类 Unix 系统中最古老的 IPC 方式之一。简单来说，它将一个进程的输出连接到另一个进程的输入，形成一条数据流通道。
关键特性：
- 单向通信：只能单工传输，双向通信需创建两个管道。
- 分类：分为匿名管道和命名管道。
2. 匿名管道
匿名管道（pipe）主要用于有亲缘关系的进程之间（如父子进程）。它本质上是内核管理的一块内存区域，通过一对文件描述符（fd[0] 读端，fd[1] 写端）来操作。随着进程退出，管道自动销毁。
核心 API
使用 pipe() 系统调用创建管道。成功返回 0，失败返回 -1。
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/types.h>
#include <unistd.h>
#include <sys/wait.h>
int main() {
int pipefd[2] = {0};
// 创建管道，pipefd[0] 为读端，pipefd[1] 为写端
if (pipe(pipefd) != 0) {
perror("pipe error");
exit(1);
}
printf("pipefd[0]: %d, pipefd[1]: %d\n", pipefd[0], pipefd[1]);
pid_t id = fork();
if (id == 0) {
// 子进程：关闭读端，只负责写
close(pipefd[0]);
char* msg = "hello pipe";
int cnt = 5;
char outbuffer[256];
while (cnt) {
snprintf(outbuffer, sizeof(outbuffer), "子->父# %s %d", msg, cnt--);
write(pipefd[1], outbuffer, strlen(outbuffer));
sleep(1);
}
close(pipefd[1]);
exit(0);
} else {
// 父进程：关闭写端，只负责读
close(pipefd[1]);
char inbuffer[1024];
while (1) {
inbuffer[0] = 0;
ssize_t n = read(pipefd[0], inbuffer, sizeof(inbuffer) - 1);
if (n > 0) {
inbuffer[n] = 0; // 手动添加结束符
printf("%s\n", inbuffer);
} else if (n == 0) {
printf("管道读取结束\n");
close(pipefd[0]);
break;
} else {
perror("read");
break;
}
}
waitpid(id, NULL, 0);
}
return 0;
}
运行逻辑说明：
- 子进程写入时，如果管道满了会阻塞；父进程读取慢也会阻塞子进程。
- 当所有写端关闭后，读端读到 EOF（返回 0），此时应关闭读端并退出循环。
- 注意：读写端都要及时关闭不用的那一端，否则会导致死锁或无法检测到 EOF。
3. 命名管道
匿名管道受限于血缘关系。如果需要无关进程通信，可以使用命名管道（FIFO）。它在文件系统中有一个路径名，任何知道该路径的进程都能打开它。
创建与管理
- 命令行创建：
mkfifo 文件名 - 程序创建：使用
mkfifo(const char *pathname, mode_t mode) - 删除：使用
unlink("文件名")清理残留文件。
命名管道的语义与匿名管道基本一致，区别仅在于创建方式和持久化路径。
三、实战：匿名管道实现进程池
进程池是一种优化手段，提前创建好多个子进程，任务到来时直接分配，避免频繁 fork 的开销。
这里我们用一个 C++ 类封装进程池，利用匿名管道作为任务分发通道。
#include <cassert>
#include <cstdio>
#include <cstdlib>
#include <ctime>
#include <iostream>
#include <string>
#include <sys/types.h>
#include <sys/wait.h>
#include <unistd.h>
#include <vector>
// 错误码定义
enum { OK = 0, PIPE_ERR, FORK_ERR, READ_ERR, WRITE_ERR, WAIT_ERR };
const int gprocessnum = 7; // 子进程数量
void task1() { std::cout << "这是下载数据任务" << std::endl; }
void task2() { std::cout << "这是打印日志任务" << std::endl; }
void task3() { std::cout << "这是刷新磁盘任务" << std::endl; }
void task4() { std::cout << "这是更新用户状态任务" << std::endl; }
typedef void (*task_t)();
const int gtasknum = 4;
task_t tasks[gtasknum] = {task1, task2, task3, task4};
class ProcessPool {
private:
class Channel {
private:
int _wfd; // 写端 fd
pid_t _id; // 子进程 ID
std::string channel_name;
public:
Channel(int wfd, pid_t id) : _wfd(wfd), _id(id) {
channel_name = "channel-" + std::to_string(id);
}
void ClosePipe() { close(_wfd); }
void PrintInfo() {
printf("管道 wfd: %d, 通道名：%s\n", _wfd, channel_name.c_str());
}
int getfd() { return _wfd; }
const char* getname() { return channel_name.c_str(); }
void Wait() {
pid_t rid = waitpid(_id, nullptr, 0);
if (rid < 0) exit(WAIT_ERR);
std::cout << "回收子进程：" << _id << std::endl;
}
};
public:
ProcessPool() { srand(time(NULL)); }
void Init() { CreateProcessChannels(); }
void Debug() {
for (auto& channel : channels) channel.PrintInfo();
}
void Run() {
int cnt = 10;
while (cnt--) {
int itask = SelectTask();
int ichannel = SelectChannel();
printf("向通道%s发送任务 task%d...\n", channels[ichannel].getname(), itask + 1);
SendTask2Channel(itask, ichannel);
sleep(1);
}
}
void Quit() {
for (auto& channel : channels) {
channel.ClosePipe();
channel.Wait();
}
}
private:
void CreateProcessChannels() {
for (int i = 0; i < gprocessnum; i++) {
int pipefd[2] = {0};
if (pipe(pipefd) < 0) exit(PIPE_ERR);
pid_t id = fork();
if (id < 0) exit(FORK_ERR);
else if (id == 0) {
// 子进程逻辑
// 注意：子进程继承父进程 fd 表，必须关闭其他通道的写端
if (!channels.empty()) {
for (auto& channel : channels) channel.ClosePipe();
}
close(pipefd[1]); // 关闭写端，只读
DoTask(pipefd[0]);
exit(OK);
} else {
// 父进程逻辑
close(pipefd[0]); // 关闭读端，只写
channels.emplace_back(pipefd[1], id);
printf("创建子进程%d成功\n", id);
}
}
}
void DoTask(int fd) {
while (1) {
int task_code;
ssize_t n = read(fd, &task_code, sizeof(task_code));
if (n == sizeof(task_code)) {
if (task_code >= 0 && task_code < gtasknum) {
tasks[task_code]();
}
} else if (n == 0) {
printf("%d任务退出\n", getpid());
break;
} else {
exit(READ_ERR);
}
}
}
int SelectTask() { return rand() % gtasknum; }
int SelectChannel() {
static int i = 0;
int selected = i;
i++;
i %= gprocessnum;
return selected;
}
void SendTask2Channel(int itask, int ichannel) {
assert(0 <= itask && itask < gtasknum && ichannel >= 0 && ichannel < gprocessnum);
ssize_t n = write(channels[ichannel].getfd(), &itask, sizeof(itask));
if (n < 0) exit(WRITE_ERR);
}
private:
std::vector<Channel> channels;
};
int main() {
ProcessPool pp;
pp.Init();
pp.Debug();
pp.Run();
pp.Quit();
return 0;
}
设计要点：
- 每个子进程维护一个独立的管道通道。
- 父进程轮询选择通道发送任务，子进程阻塞等待任务码。
- 任务执行完成后，子进程继续监听，直到父进程关闭管道。
四、实战：命名管道实现服务端客户端模型
对于无血缘关系的进程，我们可以用命名管道搭建简单的 C/S 架构。
1. 管道工具类 (Fifo.hpp)
封装了 mkfifo、open、unlink 等操作，方便复用。
// Fifo.hpp
#pragma once
#include <cstdio>
#include <cstring>
#include <fcntl.h>
#include <iostream>
#include <string>
#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>
#define FOR_READ 1
#define FOR_WRITE 2
const std::string myfifo = "./fifo";
class Fifo {
public:
Fifo(const std::string& filename = myfifo) : _filename(filename), _mode(0666), _fd(-1) {}
void Build() {
if (IsExist()) return;
int n = mkfifo(_filename.c_str(), _mode);
if (n < 0) {
std::cerr << "mkfifo error: " << strerror(errno) << std::endl;
exit(1);
}
std::cout << "mkfifo success" << std::endl;
}
void Open(int mode) {
if (mode == FOR_READ) _fd = open(_filename.c_str(), O_RDONLY);
else if (mode == FOR_WRITE) _fd = open(_filename.c_str(), O_WRONLY);
if (_fd < 0) {
std::cerr << "open error: " << strerror(errno) << std::endl;
exit(2);
}
std::cout << "open success" << std::endl;
}
void Delete() {
if (!IsExist()) return;
int n = unlink(_filename.c_str());
if (n < 0) {
std::cerr << "delete error: " << strerror(errno) << std::endl;
exit(3);
}
std::cout << "delete success" << std::endl;
}
void Send(std::string& msgin) {
write(_fd, msgin.c_str(), msgin.size());
}
int Receive(std::string& msgout) {
char buffer[128];
ssize_t n = read(_fd, buffer, sizeof(buffer) - 1);
if (n > 0) {
buffer[n] = '\0';
msgout = buffer;
return n;
} else if (n == 0) return 0;
else return -1;
}
private:
bool IsExist() {
struct stat st;
int n = stat(_filename.c_str(), &st);
return (n == 0);
}
private:
std::string _filename;
mode_t _mode;
int _fd;
};
2. 服务端 (Server.cc)
服务端先创建并打开管道（读模式），阻塞等待客户端连接。
// Server.cc
#include "Fifo.hpp"
int main() {
Fifo pipefile;
pipefile.Build();
pipefile.Open(FOR_READ);
std::string msg;
while (1) {
int n = pipefile.Receive(msg);
if (n > 0) {
std::cout << "客户端说：" << msg << std::endl;
} else {
break;
}
}
pipefile.Delete();
return 0;
}
3. 客户端 (Client.cc)
客户端打开管道（写模式），持续发送消息。
// Client.cc
#include "Fifo.hpp"
int main() {
Fifo fileclient;
fileclient.Open(FOR_WRITE);
while (1) {
std::cout << "请输入：" << std::endl;
std::string msg;
std::getline(std::cin, msg);
fileclient.Send(msg);
}
return 0;
}
注意事项：
- 命名管道打开时，读端打开可能会阻塞，直到有写端打开；反之亦然。
- 使用完毕后务必
unlink清理文件，避免残留影响后续测试。
管道通信是 Linux 系统编程的基石。理解其底层机制，有助于我们在高并发场景下做出更合理的架构选型。