深入理解C++管道編程
更新時(shí)間：2026年02月07日 10:38:08 作者：碼事漫談
本文主要介紹了深入理解C++管道編程,文中通過示例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧
第一章：管道編程的核心概念
1.1 什么是管道？
管道是UNIX和類UNIX系統(tǒng)中最古老、最基礎(chǔ)的進(jìn)程間通信（IPC）機(jī)制之一。你可以將它想象成現(xiàn)實(shí)世界中的水管：數(shù)據(jù)像水流一樣從一個(gè)進(jìn)程"流"向另一個(gè)進(jìn)程。
核心特征：
- 半雙工通信：數(shù)據(jù)只能單向流動(dòng)（要么從A到B，要么從B到A）
- 字節(jié)流導(dǎo)向：沒有消息邊界，數(shù)據(jù)是連續(xù)的字節(jié)流
- 基于文件描述符：使用與文件操作相同的接口
- 內(nèi)核緩沖區(qū)：數(shù)據(jù)在內(nèi)核緩沖區(qū)中暫存
1.2 管道的工作原理
讓我們通過一個(gè)簡單的比喻來理解管道的工作原理：
想象兩個(gè)進(jìn)程要通過管道通信：
進(jìn)程A（寫端） → [內(nèi)核緩沖區(qū)] → 進(jìn)程B（讀端）
內(nèi)核緩沖區(qū)的作用：
- 當(dāng)進(jìn)程A寫入數(shù)據(jù)時(shí)，數(shù)據(jù)先進(jìn)入內(nèi)核緩沖區(qū)
- 進(jìn)程B從緩沖區(qū)讀取數(shù)據(jù)
- 如果緩沖區(qū)空，讀操作會(huì)阻塞（等待數(shù)據(jù)）
- 如果緩沖區(qū)滿，寫操作會(huì)阻塞（等待空間）
匿名管道的關(guān)鍵限制：
- 只能用于有"親緣關(guān)系"的進(jìn)程間通信（通常是父子進(jìn)程或兄弟進(jìn)程）
- 生命周期隨進(jìn)程結(jié)束而結(jié)束
- 無法在無關(guān)進(jìn)程間使用
第二章：入門實(shí)踐——創(chuàng)建第一個(gè)管道
2.1 理解文件描述符
在深入代碼之前，必須理解文件描述符的概念：
// 每個(gè)進(jìn)程都有這三個(gè)標(biāo)準(zhǔn)文件描述符：
// 0 - 標(biāo)準(zhǔn)輸入（stdin） → 通常從鍵盤讀取
// 1 - 標(biāo)準(zhǔn)輸出（stdout） → 通常輸出到屏幕
// 2 - 標(biāo)準(zhǔn)錯(cuò)誤（stderr） → 錯(cuò)誤信息輸出// 當(dāng)創(chuàng)建管道時(shí)，系統(tǒng)會(huì)分配兩個(gè)新的文件描述符：
// pipefd[0] - 用于讀取的端
// pipefd[1] - 用于寫入的端
2.2 創(chuàng)建第一個(gè)管道程序
讓我們從最簡單的例子開始：
#include <iostream>
#include <unistd.h> // pipe(), fork(), read(), write()
#include <string.h> // strlen()
#include <sys/wait.h> // wait()
int main() {
int pipefd[2]; // 管道文件描述符數(shù)組
char buffer[100];
// 步驟1：創(chuàng)建管道
// pipe() 返回0表示成功，-1表示失敗
if (pipe(pipefd) == -1) {
std::cerr << "管道創(chuàng)建失敗" << std::endl;
return 1;
}
// 步驟2：創(chuàng)建子進(jìn)程
pid_t pid = fork();
if (pid == -1) {
std::cerr << "進(jìn)程創(chuàng)建失敗" << std::endl;
return 1;
}
if (pid == 0) {
// 子進(jìn)程代碼
// 關(guān)閉不需要的寫端
close(pipefd[1]);
// 從管道讀取數(shù)據(jù)
int bytes_read = read(pipefd[0], buffer, sizeof(buffer));
if (bytes_read > 0) {
std::cout << "子進(jìn)程收到: " << buffer << std::endl;
}
close(pipefd[0]);
return 0;
} else {
// 父進(jìn)程代碼
// 關(guān)閉不需要的讀端
close(pipefd[0]);
const char* message = "Hello from parent!";
// 向管道寫入數(shù)據(jù)
write(pipefd[1], message, strlen(message));
// 關(guān)閉寫端，表示數(shù)據(jù)發(fā)送完畢
close(pipefd[1]);
// 等待子進(jìn)程結(jié)束
wait(nullptr);
}
return 0;
}
2.3 關(guān)鍵原理分析
為什么需要關(guān)閉不用的描述符？
- 資源管理：每個(gè)進(jìn)程都有文件描述符限制，及時(shí)關(guān)閉避免泄漏
- 正確終止：讀進(jìn)程需要知道何時(shí)沒有更多數(shù)據(jù)
- 所有寫端關(guān)閉 → 讀端返回0（EOF）
- 否則讀端會(huì)一直等待
管道的阻塞行為：
- 讀阻塞：當(dāng)管道空且仍有寫端打開時(shí)，讀操作會(huì)阻塞
- 寫阻塞：當(dāng)管道滿（默認(rèn)64KB），寫操作會(huì)阻塞
- 非阻塞模式：可以通過fcntl()設(shè)置O_NONBLOCK
第三章：中級應(yīng)用——雙向通信與復(fù)雜管道
3.1 實(shí)現(xiàn)雙向通信
單個(gè)管道只能單向通信，要實(shí)現(xiàn)雙向通信，我們需要兩個(gè)管道：
#include <iostream>
#include <unistd.h>
#include <string>
class BidirectionalPipe {
private:
int parent_to_child[2]; // 父→子管道
int child_to_parent[2]; // 子→父管道
public:
BidirectionalPipe() {
// 創(chuàng)建兩個(gè)管道
if (pipe(parent_to_child) == -1 || pipe(child_to_parent) == -1) {
throw std::runtime_error("管道創(chuàng)建失敗");
}
}
~BidirectionalPipe() {
closeAll();
}
void parentWrite(const std::string& message) {
write(parent_to_child[1], message.c_str(), message.length());
}
std::string parentRead() {
char buffer[256];
int n = read(child_to_parent[0], buffer, sizeof(buffer)-1);
if (n > 0) {
buffer[n] = '\0';
return std::string(buffer);
}
return "";
}
void childWrite(const std::string& message) {
write(child_to_parent[1], message.c_str(), message.length());
}
std::string childRead() {
char buffer[256];
int n = read(parent_to_child[0], buffer, sizeof(buffer)-1);
if (n > 0) {
buffer[n] = '\0';
return std::string(buffer);
}
return "";
}
void closeParentSide() {
close(parent_to_child[1]); // 關(guān)閉父進(jìn)程的寫端
close(child_to_parent[0]); // 關(guān)閉父進(jìn)程的讀端
}
void closeChildSide() {
close(parent_to_child[0]); // 關(guān)閉子進(jìn)程的讀端
close(child_to_parent[1]); // 關(guān)閉子進(jìn)程的寫端
}
private:
void closeAll() {
close(parent_to_child[0]);
close(parent_to_child[1]);
close(child_to_parent[0]);
close(child_to_parent[1]);
}
};
3.2 管道鏈的實(shí)現(xiàn)
管道鏈?zhǔn)荱NIX shell中|操作符的基礎(chǔ)，讓我們實(shí)現(xiàn)一個(gè)簡單的版本：
#include <vector>
#include <array>
class Pipeline {
private:
// 存儲(chǔ)多個(gè)命令
std::vector<std::vector<std::string>> commands;
public:
void addCommand(const std::vector<std::string>& cmd) {
commands.push_back(cmd);
}
void execute() {
std::vector<int> prev_pipe_read; // 前一個(gè)管道的讀端
for (size_t i = 0; i < commands.size(); ++i) {
int pipefd[2];
// 如果不是最后一個(gè)命令，創(chuàng)建管道
if (i < commands.size() - 1) {
if (pipe(pipefd) == -1) {
throw std::runtime_error("管道創(chuàng)建失敗");
}
}
pid_t pid = fork();
if (pid == 0) {
// 子進(jìn)程代碼
// 設(shè)置輸入重定向（從上一個(gè)管道讀?。?
if (!prev_pipe_read.empty()) {
dup2(prev_pipe_read[0], STDIN_FILENO);
close(prev_pipe_read[0]);
}
// 設(shè)置輸出重定向（寫入下一個(gè)管道）
if (i < commands.size() - 1) {
dup2(pipefd[1], STDOUT_FILENO);
close(pipefd[0]);
close(pipefd[1]);
}
// 準(zhǔn)備exec參數(shù)
std::vector<char*> args;
for (const auto& arg : commands[i]) {
args.push_back(const_cast<char*>(arg.c_str()));
}
args.push_back(nullptr);
// 執(zhí)行命令
execvp(args[0], args.data());
// exec失敗才執(zhí)行到這里
exit(1);
} else {
// 父進(jìn)程代碼
// 關(guān)閉不再需要的描述符
if (!prev_pipe_read.empty()) {
close(prev_pipe_read[0]);
}
if (i < commands.size() - 1) {
close(pipefd[1]); // 父進(jìn)程不需要寫端
prev_pipe_read = {pipefd[0]}; // 保存讀端用于下一個(gè)進(jìn)程
}
}
}
// 父進(jìn)程等待所有子進(jìn)程
for (size_t i = 0; i < commands.size(); ++i) {
wait(nullptr);
}
}
};
// 使用示例
int main() {
Pipeline pipeline;
// 模擬: ls -l | grep ".cpp" | wc -l
pipeline.addCommand({"ls", "-l"});
pipeline.addCommand({"grep", "\\.cpp"});
pipeline.addCommand({"wc", "-l"});
pipeline.execute();
return 0;
}
3.3 命名管道（FIFO）的深入理解
命名管道與匿名管道的區(qū)別：
| 特性 | 匿名管道 | 命名管道（FIFO） |
|---|---|---|
| 持久性 | 進(jìn)程結(jié)束即消失 | 文件系統(tǒng)中有實(shí)體文件 |
| 進(jìn)程關(guān)系 | 必須有親緣關(guān)系 | 任意進(jìn)程都可訪問 |
| 創(chuàng)建方式 | pipe()系統(tǒng)調(diào)用 | mkfifo()函數(shù) |
| 訪問控制 | 基于文件描述符繼承 | 基于文件權(quán)限 |
創(chuàng)建和使用命名管道：
#include <iostream>
#include <fcntl.h>
#include <sys/stat.h>
#include <unistd.h>
class NamedPipe {
private:
std::string path;
int fd;
public:
NamedPipe(const std::string& pipePath) : path(pipePath) {
// 創(chuàng)建命名管道（如果不存在）
if (mkfifo(path.c_str(), 0666) == -1) {
// 如果已存在，忽略EEXIST錯(cuò)誤
if (errno != EEXIST) {
throw std::runtime_error("無法創(chuàng)建命名管道");
}
}
}
// 作為讀取者打開
void openForReading(bool nonblock = false) {
int flags = O_RDONLY;
if (nonblock) flags |= O_NONBLOCK;
fd = open(path.c_str(), flags);
if (fd == -1) {
throw std::runtime_error("無法打開命名管道進(jìn)行讀取");
}
}
// 作為寫入者打開
void openForWriting(bool nonblock = false) {
int flags = O_WRONLY;
if (nonblock) flags |= O_NONBLOCK;
fd = open(path.c_str(), flags);
if (fd == -1) {
throw std::runtime_error("無法打開命名管道進(jìn)行寫入");
}
}
// 讀取數(shù)據(jù)
std::string readData(size_t max_size = 1024) {
char buffer[max_size];
ssize_t bytes = read(fd, buffer, max_size - 1);
if (bytes > 0) {
buffer[bytes] = '\0';
return std::string(buffer);
}
return "";
}
// 寫入數(shù)據(jù)
void writeData(const std::string& data) {
write(fd, data.c_str(), data.length());
}
~NamedPipe() {
if (fd != -1) {
close(fd);
}
// 可以選擇是否刪除管道文件
// unlink(path.c_str());
}
};
第四章：高級主題——性能與并發(fā)
4.1 非阻塞管道操作
非阻塞管道在某些場景下非常有用，比如同時(shí)監(jiān)控多個(gè)管道：
#include <fcntl.h>
class NonBlockingPipe {
private:
int pipefd[2];
public:
NonBlockingPipe() {
if (pipe(pipefd) == -1) {
throw std::runtime_error("管道創(chuàng)建失敗");
}
// 設(shè)置為非阻塞模式
setNonBlocking(pipefd[0]);
setNonBlocking(pipefd[1]);
}
private:
void setNonBlocking(int fd) {
int flags = fcntl(fd, F_GETFL, 0);
if (flags == -1) {
throw std::runtime_error("獲取文件狀態(tài)失敗");
}
if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) == -1) {
throw std::runtime_error("設(shè)置非阻塞模式失敗");
}
}
public:
// 非阻塞讀取
bool tryRead(std::string& result) {
char buffer[1024];
ssize_t bytes = read(pipefd[0], buffer, sizeof(buffer) - 1);
if (bytes > 0) {
buffer[bytes] = '\0';
result = buffer;
return true;
} else if (bytes == -1 && errno == EAGAIN) {
// 沒有數(shù)據(jù)可讀（非阻塞模式）
return false;
}
return false; // 錯(cuò)誤或EOF
}
};
4.2 使用select實(shí)現(xiàn)多路復(fù)用
當(dāng)需要同時(shí)監(jiān)控多個(gè)管道時(shí)，select是一個(gè)非常有效的工具：
#include <sys/select.h>
#include <vector>
class PipeMonitor {
private:
std::vector<int> read_fds; // 需要監(jiān)控的讀描述符
public:
void addPipe(int read_fd) {
read_fds.push_back(read_fd);
}
// 監(jiān)控所有管道，返回有數(shù)據(jù)可讀的管道列表
std::vector<int> monitor(int timeout_sec = 0) {
fd_set read_set;
FD_ZERO(&read_set);
int max_fd = 0;
for (int fd : read_fds) {
FD_SET(fd, &read_set);
if (fd > max_fd) max_fd = fd;
}
struct timeval timeout;
timeout.tv_sec = timeout_sec;
timeout.tv_usec = 0;
// 使用select等待數(shù)據(jù)
int ready = select(max_fd + 1, &read_set, nullptr, nullptr,
timeout_sec >= 0 ? &timeout : nullptr);
std::vector<int> ready_fds;
if (ready > 0) {
for (int fd : read_fds) {
if (FD_ISSET(fd, &read_set)) {
ready_fds.push_back(fd);
}
}
}
return ready_fds;
}
};
4.3 零拷貝技術(shù)：splice()
Linux提供了高級的系統(tǒng)調(diào)用來優(yōu)化管道性能，避免不必要的數(shù)據(jù)拷貝：
#include <fcntl.h>
class HighPerformancePipe {
private:
int pipefd[2];
public:
HighPerformancePipe() {
if (pipe(pipefd) == -1) {
throw std::runtime_error("管道創(chuàng)建失敗");
}
}
// 使用splice實(shí)現(xiàn)零拷貝數(shù)據(jù)傳輸
// 將數(shù)據(jù)從一個(gè)文件描述符直接移動(dòng)到管道
ssize_t transferFrom(int source_fd, size_t len) {
// splice從source_fd讀取數(shù)據(jù)，直接寫入管道
// 避免了用戶空間的內(nèi)存拷貝
return splice(source_fd, nullptr, // 源文件描述符
pipefd[1], nullptr, // 目標(biāo)管道寫端
len, // 傳輸長度
SPLICE_F_MOVE | SPLICE_F_MORE);
}
// 將數(shù)據(jù)從管道直接傳輸?shù)侥繕?biāo)文件描述符
ssize_t transferTo(int dest_fd, size_t len) {
return splice(pipefd[0], nullptr, // 源管道讀端
dest_fd, nullptr, // 目標(biāo)文件描述符
len,
SPLICE_F_MOVE | SPLICE_F_MORE);
}
};
第五章：最佳實(shí)踐與錯(cuò)誤處理
5.1 RAII包裝器
為了避免資源泄漏，使用RAII（資源獲取即初始化）模式管理管道：
#include <memory>
class PipeRAII {
private:
int pipefd[2];
bool valid;
public:
PipeRAII() : valid(false) {
if (pipe(pipefd) == 0) {
valid = true;
}
}
~PipeRAII() {
if (valid) {
close(pipefd[0]);
close(pipefd[1]);
}
}
// 刪除拷貝構(gòu)造函數(shù)和賦值運(yùn)算符
PipeRAII(const PipeRAII&) = delete;
PipeRAII& operator=(const PipeRAII&) = delete;
// 允許移動(dòng)語義
PipeRAII(PipeRAII&& other) noexcept
: pipefd{other.pipefd[0], other.pipefd[1]},
valid(other.valid) {
other.valid = false;
}
int readEnd() const { return valid ? pipefd[0] : -1; }
int writeEnd() const { return valid ? pipefd[1] : -1; }
explicit operator bool() const { return valid; }
};
// 使用智能指針管理
class SafePipeManager {
private:
std::unique_ptr<PipeRAII> pipe;
public:
SafePipeManager() : pipe(std::make_unique<PipeRAII>()) {
if (!*pipe) {
throw std::runtime_error("管道創(chuàng)建失敗");
}
}
void sendData(const std::string& data) {
if (pipe) {
write(pipe->writeEnd(), data.c_str(), data.length());
}
}
};
5.2 常見錯(cuò)誤與處理
class RobustPipe {
private:
int pipefd[2];
// 安全讀取函數(shù)
ssize_t safeRead(void* buf, size_t count) {
ssize_t bytes_read;
do {
bytes_read = read(pipefd[0], buf, count);
} while (bytes_read == -1 && errno == EINTR); // 處理信號中斷
return bytes_read;
}
// 安全寫入函數(shù)
ssize_t safeWrite(const void* buf, size_t count) {
ssize_t bytes_written;
size_t total_written = 0;
const char* ptr = static_cast<const char*>(buf);
while (total_written < count) {
do {
bytes_written = write(pipefd[1], ptr + total_written,
count - total_written);
} while (bytes_written == -1 && errno == EINTR);
if (bytes_written == -1) {
// 處理真正的錯(cuò)誤
if (errno == EPIPE) {
std::cerr << "管道斷裂：讀端已關(guān)閉" << std::endl;
}
return -1;
}
total_written += bytes_written;
}
return total_written;
}
public:
RobustPipe() {
if (pipe(pipefd) == -1) {
// 檢查具體錯(cuò)誤
switch (errno) {
case EMFILE:
throw std::runtime_error("進(jìn)程文件描述符耗盡");
case ENFILE:
throw std::runtime_error("系統(tǒng)文件描述符耗盡");
default:
throw std::runtime_error("未知管道創(chuàng)建錯(cuò)誤");
}
}
// 設(shè)置管道緩沖區(qū)大小（可選）
int size = 65536; // 64KB
fcntl(pipefd[0], F_SETPIPE_SZ, size);
}
};
第六章：實(shí)戰(zhàn)應(yīng)用案例
6.1 日志收集系統(tǒng)
#include <thread>
#include <queue>
#include <mutex>
#include <condition_variable>
class LogCollector {
private:
int log_pipe[2];
std::queue<std::string> log_queue;
std::mutex queue_mutex;
std::condition_variable queue_cv;
std::thread worker_thread;
bool running;
void worker() {
char buffer[4096];
while (running) {
ssize_t bytes = read(log_pipe[0], buffer, sizeof(buffer) - 1);
if (bytes > 0) {
buffer[bytes] = '\0';
std::string log_entry(buffer);
{
std::lock_guard<std::mutex> lock(queue_mutex);
log_queue.push(log_entry);
}
queue_cv.notify_one();
}
}
}
public:
LogCollector() : running(true) {
if (pipe(log_pipe) == -1) {
throw std::runtime_error("日志管道創(chuàng)建失敗");
}
worker_thread = std::thread(&LogCollector::worker, this);
}
~LogCollector() {
running = false;
close(log_pipe[1]); // 關(guān)閉寫端，使讀端退出
if (worker_thread.joinable()) {
worker_thread.join();
}
close(log_pipe[0]);
}
// 寫入日志
void log(const std::string& message) {
write(log_pipe[1], message.c_str(), message.length());
}
// 獲取日志（線程安全）
std::string getLog() {
std::unique_lock<std::mutex> lock(queue_mutex);
queue_cv.wait(lock, [this] { return !log_queue.empty(); });
std::string log = log_queue.front();
log_queue.pop();
return log;
}
};
總結(jié)
管道編程是C++系統(tǒng)編程的重要部分，掌握它需要：
- 理解基本原理：文件描述符、緩沖區(qū)、阻塞行為
- 掌握核心API：pipe(), fork(), dup2(), read(), write()
- 學(xué)會(huì)高級技術(shù)：非阻塞IO、多路復(fù)用、零拷貝
- 遵循最佳實(shí)踐：RAII管理、錯(cuò)誤處理、資源清理
管道不僅是一種技術(shù)，更是一種設(shè)計(jì)哲學(xué)——它鼓勵(lì)我們創(chuàng)建模塊化、可組合的程序，這正是UNIX哲學(xué)的核心理念之一。
到此這篇關(guān)于深入理解C++管道編程的文章就介紹到這了,更多相關(guān)C++管道編程內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
本文主要介紹C語言簡單的程序,這里給大家詳細(xì)介紹C語言代碼,對函數(shù)概念、頭文件、自定義函數(shù)等基礎(chǔ)信息的講解,希望能幫助剛剛學(xué)習(xí)的同學(xué)2016-07-07
這篇文章主要為大家詳細(xì)介紹了C++實(shí)現(xiàn)馬踏棋盤，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2022-02-02
本篇文章是對C語言中空指針、空指針常量、NULL & 0 進(jìn)行了詳細(xì)的分析介紹，需要的朋友參考下2013-05-05
這篇文章主要為大家詳細(xì)介紹了C語言實(shí)現(xiàn)掃雷游戲初級版，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2021-08-08
這篇文章主要介紹了C++ vector的基本使用,本文通過示例代碼給大家介紹的非常詳細(xì)，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下2023-03-03
C語言中pthread_exit()函數(shù)實(shí)現(xiàn)終止線程
本文主要介紹了C語言中pthread_exit()函數(shù)實(shí)現(xiàn)終止線程，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2023-05-05
這篇文章主要為大家詳細(xì)介紹了C語言開發(fā)實(shí)現(xiàn)掃雷游戲，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2020-11-11
C++中的三種繼承public,protected,private詳細(xì)解析
我們已經(jīng)知道，在基類以private方式被繼承時(shí)，其public和protected成員在子類中變?yōu)閜rivate成員。然而某些情況下，需要在子類中將一個(gè)或多個(gè)繼承的成員恢復(fù)其在基類中的訪問權(quán)限2013-09-09
這篇文章主要介紹了C++事件處理中__hook與__unhook的用法,C++中的COM類主要支持事件處理,需要的朋友可以參考下2016-01-01
解析bitmap處理海量數(shù)據(jù)及其實(shí)現(xiàn)方法分析
本篇文章是對bitmap處理海量數(shù)據(jù)及其實(shí)現(xiàn)的方法進(jìn)行了詳細(xì)的分析介紹，需要的朋友參考下2013-05-05