流式管道与命名管道吞吐
使用 FIFO 和进程替换在各阶段之间流式传输数据，无需中间文件。
流式管道与命名管道吞吐 是 CoddyKit 上的免费 DevOps Bootcamp 课时。 这是第 4 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 DevOps Bootcamp 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 DevOps Bootcamp 课程共包含 4 节课。
为什么中间文件会降低吞吐量
当您使用 sort file.txt > tmp.txt && uniq tmp.txt > result.txt 这样的方式串联命令时，需要承担一项隐藏成本：磁盘写入、磁盘读取，以及流水线必须等第一阶段完全结束后下一阶段才能开始，因此会发生停顿。
流式流水线可以消除这项成本。数据在内存中直接从生产者流向消费者，并按阶段并发处理。这正是 Unix 管道背后的核心思想，而命名管道（FIFO）进一步扩展了这一思想。
- 匿名管道（
|）：连接同一 Shell 命令行中相邻的两个命令。 - 命名管道（FIFO）：文件系统中的特殊文件，使互不相关的进程能够相互传输数据流。
- 进程替换：让一个命令可以将另一个命令的输出当作文件使用。
本课将展示如何在实际 Bash 工作流中应用这三种方式，以最大化吞吐量。
流式流水线的结构
匿名管道将一个进程的标准输出连接到下一个进程的标准输入。内核会让两个进程同时运行，并使用固定大小的内存缓冲区（在 Linux 上通常为 64 KB）。
关键认识是：流水线的速度取决于最慢的阶段。如果生产者更快，它会因缓冲区已满而阻塞；如果消费者更快，它会因缓冲区为空而阻塞。这种反压是一种免费且自动的流量控制。
下面的示例会统计大型访问日志中的不重复 IP 地址，整个过程从不写入临时文件。每个阶段都会并发运行：
#!/usr/bin/env bash
# Stream a 2 GB access log — all stages run in parallel
grep '"GET' /var/log/nginx/access.log \
| awk '{print $1}' \
| sort \
| uniq -c \
| sort -rn \
| head -20使用 mkfifo 创建命名管道
使用 mkfifo 可以创建命名管道（FIFO — 先进先出）。它在文件系统中看起来像普通文件，但写入其中的数据不会存储在磁盘上，而是直接流向读取进程。
请记住以下关键行为：
- 向 FIFO 写入时，会一直阻塞，直到有读取者打开它；反之亦然。
- FIFO 条目会保留在文件系统中；使用完毕后必须用
rm将其删除。 - 允许有多个写入者，但无法保证它们之间的顺序。
下面的示例中，生产者会将数据压缩到 FIFO 中，同时消费者将其上传到 S3，不需要临时文件。
#!/usr/bin/env bash
mkfifo /tmp/stream_pipe
# Producer: compress in background
gzip -c /var/log/syslog > /tmp/stream_pipe &
# Consumer: read from FIFO (runs in foreground)
wc -l < /tmp/stream_pipe
wait
rm /tmp/stream_pipe进程替换：将命令当作文件使用
进程替换使用 <(command) 或 >(command) 语法。Bash 会在幕后创建 FIFO（或 /dev/fd/N 文件描述符），并将路径传递给外部命令。
当工具需要的是 文件名参数而不是标准输入时，这种方式非常有用。如果不使用进程替换，您就需要临时文件；使用它后，数据可以直接进行流式传输。
<(cmd)：外部命令从 cmd 的输出中读取。>(cmd)：外部命令向 cmd 的输入中写入。
#!/usr/bin/env bash
# diff two sorted streams without creating temp files
diff <(sort /etc/passwd) <(sort /etc/group)
# Compare live command output against a baseline
diff <(ls /usr/bin | sort) <(cat ~/bin_baseline.txt | sort)tee：将数据流拆分给多个消费者
tee 从标准输入读取内容，并将其写入标准输出以及一个或多个文件。结合进程替换，您可以将一个数据流同时分发到多个处理流水线，而完全不接触磁盘。
当您希望同时记录原始数据并对其进行处理时，这种模式非常有用。
#!/usr/bin/env bash
# Generate 100000 random numbers, then simultaneously:
# 1. compute the sum
# 2. find the maximum
# 3. count lines (saved to a variable)
seq 1 100000 \
| tee >(awk '{s+=$1} END{print "Sum:", s}') \
>(awk 'BEGIN{m=0} $1>m{m=$1} END{print "Max:", m}') \
| wc -l | xargs echo "Count:"扇出模式：一个生产者，多个消费者
当一个数据源必须为多个独立消费者提供数据时，可以将 tee 与多个 >() 进程替换结合使用。每个消费者都会获得完整的数据流，并发运行。
这样可以避免多次读取源文件。对于一个 10 GB 的文件，这种差异非常显著：只需读取磁盘一次，而不是 N 次。
#!/usr/bin/env bash
# Read a large CSV once; simultaneously:
# - count rows
# - extract column 2 to a file
# - pass column 3 to a stats script
cat large_data.csv \
| tee \
>(wc -l > /tmp/row_count.txt) \
>(cut -d',' -f2 > /tmp/col2.txt) \
>(cut -d',' -f3 | awk '{sum+=$1} END{print sum}' > /tmp/col3_sum.txt) \
> /dev/null
echo "Rows:" $(cat /tmp/row_count.txt)
echo "Col3 sum:" $(cat /tmp/col3_sum.txt)扇入模式：多个生产者，一个消费者
扇出的相反模式是扇入：多个独立数据源将数据流传入一个消费者。使用命名 FIFO 可以轻松实现这一点。
一个常见用例是实时合并多个服务器的日志流，或汇总并行工作者产生的部分结果。
请注意，存在多个写入者时，消费者看到的输出会彼此交错。对于每行自成一体的面向行数据，这通常没有问题；但如果顺序很重要，则必须自行处理排序。
#!/usr/bin/env bash
mkfifo /tmp/fanin_pipe
# Three producers write concurrently into the same FIFO
for host in web1 web2 web3; do
ssh "$host" 'tail -n 500 /var/log/app.log' > /tmp/fanin_pipe &
done
# Single consumer reads all merged output
grep 'ERROR' /tmp/fanin_pipe | sort | uniq -c | sort -rn
wait
rm /tmp/fanin_pipe使用 mkfifo 进行并行压缩
FIFO 最实用的用途之一是并行压缩。pigz（并行 gzip）或 pbzip2 等工具可以读取数据流；您可以直接传输原始数据，而无需先生成未压缩文件。
下面的模式会归档一个目录，使用全部 CPU 核心进行压缩，并将结果流式传输到远程主机，所有步骤同时进行：
#!/usr/bin/env bash
# Tar + parallel compress + stream to remote — no temp files
# Requires: pigz (parallel gzip)
tar cf - /data/large_dir \
| pigz -p 4 \
| ssh backup-host 'cat > /backups/large_dir.tar.gz'
# Verify the remote file exists
ssh backup-host 'ls -lh /backups/large_dir.tar.gz'控制缓冲区大小与阻塞
管道具有内核缓冲区（通常为 64 KB）。缓冲区已满时，写入者会阻塞；缓冲区为空时，读取者会阻塞。这通常正是您想要的行为，但在某些情况下，阻塞可能导致死锁。
死锁风险：如果进程 A 向 FIFO1 写入并从 FIFO2 读取，而进程 B 向 FIFO2 写入并从 FIFO1 读取，那么双方都可能阻塞，等待对方先完成读取。
解决方案：
- 将至少一侧置于后台（
&），使其不会阻塞 Shell。 - 使用
mbuffer或pv在各阶段之间添加更大的内存缓冲区。 - 使用
pv -q -B 128m插入一个 128 MB 的缓冲区，以平滑吞吐量峰值。
#!/usr/bin/env bash
# pv adds a 64 MB buffer and shows throughput
# Useful when producer and consumer have bursty speeds
dd if=/dev/urandom bs=1M count=200 \
| pv -B 64m \
| gzip \
| wc -c实际示例：实时日志聚合器
下面是一个完整且真实的模式：跟踪多个日志文件，通过命名管道合并数据流，筛选错误，并写入实时摘要，整个过程不使用任何中间文件，所有阶段均并行运行。
这类流水线通常会作为生产服务器上的后台监控脚本运行。
#!/usr/bin/env bash
FIFO=/tmp/log_aggregator
mkfifo "$FIFO"
cleanup() { rm -f "$FIFO"; }
trap cleanup EXIT INT TERM
# Fan-in: tail multiple logs into the FIFO
tail -F /var/log/syslog /var/log/auth.log > "$FIFO" &
TAIL_PID=$!
# Consumer: filter and timestamp errors in real time
grep --line-buffered -i 'error\|fail\|crit' "$FIFO" \
| while IFS= read -r line; do
printf '[%s] %s\n' "$(date '+%H:%M:%S')" "$line"
done
kill "$TAIL_PID" 2>/dev/null对比流水线与临时文件的基准测试
您可以使用 time 测量流式方案与临时文件方案之间真实的吞吐量差异。对于大型数据集，流水线方案更快，原因如下：
- 各阶段并发运行，CPU 与 I/O 可以重叠执行。
- 中间数据无需进行磁盘 I/O，只有最终输出会写入磁盘。
- 无论输入大小如何，内存占用都保持不变（传输的是数据流，而不是缓冲全部数据）。
下面是一个用于比较两种方案的简单基准测试：
#!/usr/bin/env bash
# Approach 1: Temp file (sequential)
time bash -c '
seq 1 5000000 > /tmp/nums.txt
sort -n /tmp/nums.txt > /tmp/sorted.txt
uniq /tmp/sorted.txt | wc -l
rm /tmp/nums.txt /tmp/sorted.txt
'
echo '---'
# Approach 2: Streaming pipeline (concurrent)
time bash -c 'seq 1 5000000 | sort -n | uniq | wc -l'知识检查：命名管道的阻塞行为
请考虑以下脚本：
mkfifo /tmp/mypipe
echo 'hello' > /tmp/mypipe
echo 'done'如果在没有任何后台进程或读取者的情况下运行此脚本，会发生什么？
回顾：流式流水线与命名管道
在本课中，您探索了如何在进程之间高效传输数据，而无需使用中间文件：
- 匿名管道（
|）连接相邻命令，并发运行所有阶段，同时自动实现反压。 - 命名管道（
mkfifo）会创建一个 FIFO 文件系统条目，使彼此无关的进程或后台进程能够相互进行流式传输——在有读取端之前，写入操作会阻塞。 - 进程替换（
<(cmd)、>(cmd)）让原本需要文件名的命令能够透明地读取或生成数据流。 tee+>()可以将一个数据流分发给多个并发消费者，而无需重新读取源数据。- 扇入通过共享 FIFO 将多个生产者合并到一个消费者。
- 反压与阻塞是功能而不是错误——但请始终至少将 FIFO 对中的一端放到后台运行，以避免死锁。
- 当各阶段以突发方式运行时，请使用
pv或mbuffer增加缓冲区并监控吞吐量。
这些技术是高吞吐量 Bash 数据工程的基础：以恒定内存处理 GB 规模的数据，并最大限度地实现 CPU/IO 并行。
常见问题解答
「流式管道与命名管道吞吐」课时是免费的吗？
是的 — 「流式管道与命名管道吞吐」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 DevOps Bootcamp 课程的其余内容，请升级到 CoddyKit PRO。 DevOps Bootcamp 课程共包含 4 节课。
「流式管道与命名管道吞吐」这节课中我会学到什么？
使用 FIFO 和进程替换在各阶段之间流式传输数据，无需中间文件。 你通过在浏览器中直接运行的动手代码来练习 DevOps Bootcamp，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 DevOps Bootcamp 需要有经验吗？
无需任何先前经验。CoddyKit 上的 DevOps Bootcamp 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 4 节课，共 4 节。
「流式管道与命名管道吞吐」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 DevOps Bootcamp 课中编写并运行代码吗？
能。每节 DevOps Bootcamp 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。