Unix 设计哲学教程 / 第 4 章：文本流与管道
第 4 章：文本流与管道
“If you can’t explain it to a computer in text, you don’t understand it well enough.”
文本流（Text Stream）和管道（Pipe）是 Unix 最具革命性的创新。它们让独立的小程序能够协同工作，形成强大的数据处理流水线。本章深入探讨这一机制的原理与实战。
4.1 文本流：Unix 的通用语言
为什么是文本？
Unix 选择纯文本作为数据交换格式，这一决策看似简单，实则深远：
| 选择纯文本的理由 | 具体好处 |
|---|---|
| 人类可读 | 可以直接用 cat、less 查看 |
| 工具兼容 | 所有 Unix 工具都能处理文本 |
| 调试友好 | 管道中任一点都可以插入调试 |
| 跨平台 | 不依赖特定的二进制格式 |
| 简单 | 不需要复杂的解析库 |
| 可组合 | 任何程序的输出都可以成为另一个程序的输入 |
文本流的行模型
Unix 文本流的基本单位是"行"（line）
每行以换行符 \n 结尾
行 1\n
行 2\n
行 3\n
EOF（输入结束）
这个简单的约定使得：
├── grep 可以逐行匹配
├── sort 可以逐行排序
├── wc 可以计数行数
└── head/tail 可以截取前/后 N 行
4.2 标准流（Standard Streams）
stdin、stdout、stderr
每个 Unix 进程启动时自动获得三个标准流：
# 查看当前进程的标准流
ls -la /proc/$$/fd/
# 0 -> /dev/pts/0 (stdin)
# 1 -> /dev/pts/0 (stdout)
# 2 -> /dev/pts/0 (stderr)
| 流 | 文件描述符 | 用途 | 默认设备 |
|---|---|---|---|
| stdin | 0 | 程序的输入 | 键盘 |
| stdout | 1 | 程序的正常输出 | 终端 |
| stderr | 2 | 程序的错误输出 | 终端 |
# stdin: 从键盘读取
cat # 等待输入，输入后回车即显示，Ctrl+D 结束
# stdout: 正常输出
echo "hello" # 输出到终端
# stderr: 错误输出
ls /nonexistent # 错误信息输出到 stderr
为什么要分离 stdout 和 stderr？
分离的好处
├── 管道只处理正常数据，不受错误信息干扰
├── 可以分别重定向到不同目标
├── 错误信息始终显示在终端，不被吞掉
└── 脚本可以分别捕获成功输出和错误信息
# 示例：将 stdout 重定向到文件，stderr 仍然显示在终端
find / -name "*.conf" > results.txt 2>/dev/null
# 将 stdout 和 stderr 分别重定向到不同文件
command > output.txt 2> error.txt
# 将 stdout 和 stderr 合并到同一个文件
command > all_output.txt 2>&1
# 或者（Bash 4+ 的简写）
command &> all_output.txt
4.3 重定向（Redirection）
基本重定向操作符
# > — 覆盖写入（stdout）
echo "hello" > file.txt
# >> — 追加写入（stdout）
echo "world" >> file.txt
# < — 从文件读取（stdin）
cat < file.txt
# 2> — 重定向 stderr
ls /nonexistent 2> error.log
# 2>> — 追加 stderr
ls /another/bad/path 2>> error.log
# 2>&1 — 将 stderr 合并到 stdout
command 2>&1 | grep "error"
# &> — 同时重定向 stdout 和 stderr（Bash）
command &> all.log
# << Here Document — 内联输入
cat << EOF
line 1
line 2
line 3
EOF
# <<< Here String — 将字符串作为 stdin
grep "pattern" <<< "search this string"
文件描述符的高级操作
# 打开自定义文件描述符
exec 3> /tmp/custom.log # 打开 fd 3 用于写入
echo "写入 fd 3" >&3 # 写入文件描述符 3
exec 3>&- # 关闭 fd 3
# 打开 fd 用于读取
exec 4< /etc/hostname
read -r hostname <&4
echo "Hostname: $hostname"
exec 4<&- # 关闭 fd 4
# 读写模式打开
exec 5<> /tmp/readwrite.txt
echo "data" >&5
read -r line <&5
exec 5>&-
# 交换 stdout 和 stderr
command 3>&1 1>&2 2>&3
重定向实战示例
# 1. 同时记录日志和显示在终端
#!/bin/bash
exec > >(tee -a /var/log/myscript.log) 2>&1
echo "This goes to both terminal and log file"
# 2. 静默执行，只在失败时显示错误
#!/bin/bash
command > /dev/null 2>&1 || {
echo "命令失败：" >&2
command 2>&1 # 重新执行，显示错误
}
# 3. 重定向到多个目标（使用 tee）
echo "hello" | tee file1.txt file2.txt file3.txt
# 4. 将命令输出保存到变量，同时显示在终端
output=$(command 2>&1 | tee /dev/tty)
# 5. 使用临时文件描述符避免子 Shell 陷阱
#!/bin/bash
while read -r line; do
echo "Processing: $line"
done < <(find /tmp -name "*.log")
# 注意 <() 是进程替换，不是简单的重定向
4.4 管道（Pipe）
管道的原理
管道是 Unix 中最优雅的进程间通信机制。| 操作符将前一个命令的 stdout 连接到后一个命令的 stdin。
管道的工作原理：
process1 process2
┌─────────┐ ┌─────────┐
│ │ stdout ──→ stdin │
│ grep │ (pipe) │ sort │
│ │ │ │
└─────────┘ └─────────┘
底层实现：
├── 内核创建一个内核缓冲区（通常 64KB）
├── 写入端（process1 的 stdout）向缓冲区写入
├── 读取端（process2 的 stdin）从缓冲区读取
├── 当缓冲区满时，写入端阻塞
├── 当缓冲区空时，读取端阻塞
└── 所有进程并行执行（不是串行）
# 管道中的进程是并行执行的
# 这个命令会立即开始输出，不需要等待 find 完成
find / -name "*.log" 2>/dev/null | head -5
# 管道的缓冲区大小
cat /proc/sys/fs/pipe-max-size # 通常 1MB
ulimit -a | grep pipe # 查看 pipe 缓冲区限制
经典管道链
# 1. 统计系统中最耗内存的 10 个进程
ps aux --sort=-%mem | head -11
# 2. 查找并删除所有 .tmp 文件
find /tmp -name "*.tmp" -type f | xargs rm -f
# 3. 统计 HTTP 状态码分布
cat access.log | awk '{print $9}' | sort | uniq -c | sort -rn
# 4. 实时监控日志
tail -f /var/log/syslog | grep --line-buffered "error"
# 5. 查找重复文件（基于 MD5）
find . -type f -exec md5sum {} + | sort | uniq -w32 -dD
# 6. 批量重命名文件
ls *.JPG | sed 's/\.JPG$/.jpg/' | while read -r new; do
mv "${new%.jpg}.JPG" "$new"
done
# 7. 文本分析：最常用的 20 个单词
cat article.txt | tr -s ' ' '\n' | tr 'A-Z' 'a-z' | sort | uniq -c | sort -rn | head -20
# 8. 多条件过滤
cat /var/log/auth.log | grep "Failed" | grep -oP '\d+\.\d+\.\d+\.\d+' | sort | uniq -c | sort -rn | head -10
4.5 tee：管道中的分流器
基本用法
tee 命令从 stdin 读取数据，同时写入 stdout 和文件。它像一个 T 形水管接头。
# 基本用法：同时输出到终端和文件
echo "hello" | tee output.txt
# 追加模式
echo "world" | tee -a output.txt
# 在管道中间使用 tee 来调试
cat data.txt | tee /dev/stderr | sort | uniq > result.txt
# 这样可以看到 sort 之前的原始数据
tee 的高级用法
# 1. 同时写入多个文件
echo "log message" | tee file1.txt file2.txt file3.txt
# 2. 管道调试：在每个步骤后查看中间结果
cat access.log \
| tee /tmp/step1_raw.txt \
| grep "ERROR" \
| tee /tmp/step2_filtered.txt \
| awk '{print $1, $4}' \
| tee /tmp/step3_extracted.txt \
| sort \
| uniq -c \
> /tmp/final_result.txt
# 3. 以 root 权限写入受限文件
echo "127.0.0.1 myhost" | sudo tee -a /etc/hosts
# 4. 使用 process substitution 将 tee 输出到命令
echo "hello" | tee >(tr 'a-z' 'A-Z') >(wc -c) > /dev/null
4.6 xargs：将 stdin 转化为参数
基本用法
xargs 从 stdin 读取数据，将它们作为参数传递给指定的命令。
# 基本：将 stdin 转为命令参数
echo "file1 file2 file3" | xargs ls -la
# 查找并删除
find /tmp -name "*.tmp" | xargs rm -f
# 与 grep 配合
find . -name "*.py" | xargs grep "import os"
xargs 的关键选项
# -I{} —— 指定替换标记
find . -name "*.log" | xargs -I{} cp {} /backup/
# -0 —— 以 null 字节分隔（处理带空格的文件名）
find . -name "*.log" -print0 | xargs -0 rm -f
# -P N —— 并行执行 N 个进程
find . -name "*.jpg" | xargs -P 4 -I{} convert {} -resize 50% thumb_{}
# -n N —— 每次传递 N 个参数
echo "a b c d e f" | xargs -n 2 echo
# 输出:
# a b
# c d
# e f
# -t —— 打印将要执行的命令（调试用）
echo "file1 file2" | xargs -t rm
# -p —— 执行前确认
echo "important_file" | xargs -p rm
# --max-args / --max-chars —— 控制参数数量和总长度
xargs vs 管道
# ❌ 不要用 for 循环处理大量文件（慢）
for f in $(find / -name "*.log"); do
rm "$f"
done
# ✅ 使用 xargs（高效，自动分批）
find / -name "*.log" -print0 | xargs -0 rm -f
# ❌ 不要简单地管道到命令参数（不会自动传递）
echo "file.txt" | rm # 错误！rm 不从 stdin 读取
# ✅ 使用 xargs
echo "file.txt" | xargs rm
4.7 管道的缓冲与死锁
管道缓冲区
# Linux 默认管道缓冲区大小
getconf PIPE_BUF / # 通常 4096 字节（原子写入大小）
# 最大管道缓冲区
cat /proc/sys/fs/pipe-max-size # 通常 1048576 字节（1MB）
管道死锁场景
管道死锁发生在：
当进程 A 向管道写入超过缓冲区大小的数据，
而进程 B 试图向另一个管道写入但那个管道的缓冲区也满了，
而进程 C 需要从第二个管道读取才能继续，
但 C 在等待第一个管道的数据...
经典的解决方案：
├── 使用更大的缓冲区
├── 使用临时文件替代过大的管道数据
└── 使用 xargs 分批处理
# 死锁示例（两个管道互相依赖）
# 假设 pipe1 的缓冲区是 64KB，pipe2 也是 64KB
# 如果 process1 需要先向 pipe2 写入 100KB 才能继续从 pipe1 读取
# 而 process2 需要先从 pipe1 读取才能向 pipe2 写入
# → 死锁！
# 实际中的规避：
# 使用临时文件存储中间结果
mkfifo pipe1 pipe2
# 安全的方式
cat input.txt | grep "pattern" > /tmp/filtered.txt
sort /tmp/filtered.txt | uniq > result.txt
rm /tmp/filtered.txt
4.8 进程替换（Process Substitution）
<() 和 >()
Bash 的进程替换特性允许将命令的输出/输入伪装成文件。
# <(command) —— 将命令输出作为"文件"
diff <(ls dir1) <(ls dir2)
# >(command) —— 将命令输入作为"文件"
echo "hello" > >(tr 'a-z' 'A-Z')
# 实际应用：
# 1. 比较两个目录的文件列表
diff <(ls /dir1 | sort) <(ls /dir2 | sort)
# 2. 比较两个命令的输出
diff <(curl -s url1) <(curl -s url2)
# 3. 同时向多个命令发送输入
cat data.txt | tee >(gzip > data.txt.gz) >(bzip2 > data.txt.bz2) > /dev/null
# 4. 避免临时文件
# 传统方式（需要临时文件）
mkfifo /tmp/pipe1 /tmp/pipe2
command1 > /tmp/pipe1 &
command2 > /tmp/pipe2 &
diff /tmp/pipe1 /tmp/pipe2
rm /tmp/pipe1 /tmp/pipe2
# 进程替换方式（无需临时文件）
diff <(command1) <(command2)
4.9 实战：构建数据处理流水线
场景：分析 Web 服务器日志
#!/bin/bash
# Nginx/Apache 日志分析工具
LOG_FILE="/var/log/nginx/access.log"
echo "=== 日志分析报告 ==="
echo "日志文件: $LOG_FILE"
echo "分析时间: $(date)"
echo ""
# 1. 总请求数
total=$(wc -l < "$LOG_FILE")
echo "总请求数: $total"
# 2. 独立 IP 数
echo "独立 IP 数: $(awk '{print $1}' "$LOG_FILE" | sort -u | wc -l)"
# 3. HTTP 状态码分布
echo ""
echo "--- HTTP 状态码分布 ---"
awk '{print $9}' "$LOG_FILE" | sort | uniq -c | sort -rn | while read -r count code; do
pct=$(echo "scale=1; $count * 100 / $total" | bc)
printf "%6s %6d (%s%%)\n" "$code" "$count" "$pct"
done
# 4. Top 10 访问 IP
echo ""
echo "--- Top 10 访问 IP ---"
awk '{print $1}' "$LOG_FILE" | sort | uniq -c | sort -rn | head -10
# 5. Top 10 请求路径
echo ""
echo "--- Top 10 请求路径 ---"
awk '{print $7}' "$LOG_FILE" | sort | uniq -c | sort -rn | head -10
# 6. 每小时请求量分布
echo ""
echo "--- 每小时请求量 ---"
awk -F'[/:]' '{print $6":"$7}' "$LOG_FILE" | sort | uniq -c | sort -k2
# 7. 4xx/5xx 错误的详细信息
echo ""
echo "--- 最近 10 条错误请求 ---"
awk '$9 >= 400' "$LOG_FILE" | tail -10
# 8. 流量统计（假设最后一列是字节数）
echo ""
echo "--- 总流量 ---"
awk '{sum+=$10} END {printf "%.2f GB\n", sum/1024/1024/1024}' "$LOG_FILE"
场景：CSV 数据处理
#!/bin/bash
# 处理 CSV 文件的 Unix 方式
DATA_FILE="sales.csv"
# 格式: 日期,产品,数量,单价
# 1. 查看前 5 行
head -5 "$DATA_FILE"
# 2. 按产品统计总销售额
tail -n +2 "$DATA_FILE" | # 跳过表头
awk -F, '{sales[$2]+=$3*$4} END {for (p in sales) printf "%s: ¥%.2f\n", p, sales[p]}' |
sort -t: -k2 -rn
# 3. 按月统计销售趋势
tail -n +2 "$DATA_FILE" |
awk -F, '{
split($1, d, "-")
month = d[1]"-"d[2]
sales[month] += $3 * $4
} END {
for (m in sales) printf "%s: ¥%.2f\n", m, sales[m]
}' |
sort
# 4. 找出销售额最高的产品
tail -n +2 "$DATA_FILE" |
awk -F, '{sales[$2]+=$3*$4} END {for (p in sales) print sales[p], p}' |
sort -rn |
head -1 |
awk '{print $2}'
# 5. 数据验证：找出异常记录
tail -n +2 "$DATA_FILE" |
awk -F, '$3 <= 0 || $4 <= 0 {print NR+1": "$0}'
场景：实时日志监控与告警
#!/bin/bash
# 实时监控日志并在发现错误时告警
LOG_FILE="/var/log/app/error.log"
ALERT_EMAIL="admin@example.com"
ERROR_THRESHOLD=5
# 使用 tail -f 实时跟踪日志
tail -f "$LOG_FILE" | while IFS= read -r line; do
# 检查是否包含错误关键词
if echo "$line" | grep -qiE "(fatal|critical|panic|oom)"; then
# 发送告警
echo "[$(date)] ALERT: $line" | mail -s "系统告警" "$ALERT_EMAIL"
logger -p local0.err "ALERT: $line"
fi
# 检查错误频率（滑动窗口）
recent_errors=$(tail -100 "$LOG_FILE" | grep -c "ERROR")
if [ "$recent_errors" -ge "$ERROR_THRESHOLD" ]; then
echo "错误频率过高: 最近100行中有 $recent_errors 条错误" |
mail -s "错误频率告警" "$ALERT_EMAIL"
fi
done
注意事项
管道中的变量作用域：管道中的每个命令在子 Shell 中执行，无法修改父 Shell 的变量。
# ❌ 错误：count 不会被修改 count=0 echo "1 2 3" | while read -r num; do ((count++)); done echo "$count" # 仍然是 0 # ✅ 正确：使用进程替换 count=0 while read -r num; do ((count++)); done < <(echo "1 2 3") echo "$count" # 3管道的退出码：默认情况下，管道的退出码是最后一个命令的退出码。使用
set -o pipefail使其成为第一个失败命令的退出码。二进制数据：管道传输的是字节流。如果需要传输二进制数据，考虑使用
base64编码或临时文件。管道中的 SIGPIPE：当读取端提前关闭（如
head取够了行数），写入端会收到 SIGPIPE 信号。