管道与过滤器

概念定义

复杂处理分解为阶段，每个阶段专注做好一种变换。管道是哑传输，过滤器是智变换；组合靠接线——同一批过滤器拼出不同管线。Unix shell 是原型，流处理器与构建系统是工业形态。

工作机制

过滤器声明输入输出模式；管道搬运数据——进程内迭代器、队列或日志主题。各阶段独立扩缩：慢过滤器的并行实例倍增其吞吐；有界管道天然传播背压。错误策略：整管线失败、单条进死信、或路由到隔离过滤器。

使用场景

步骤会演进的多步变换：ETL、文档处理、事件富化、CI 流水线。单个程序用函数就能完成同样工作时，不必付传输开销。

常见误区

“只用于流数据”是错觉：编译器、构建系统、CI 正是此形态。过滤器只该知道输入输出契约——认识邻居就破坏可重配置性。管段边界每处都付序列化与拷贝——进程内管线安全处应传引用。巨型可变对象在过滤器间穿行并被就地修改会腐蚀阶段独立性；静默吞数据的过滤器是管线里的黑洞。管线深度是成本决策：每阶段都加缓冲与故障面。

代码示例

#!/usr/bin/env bash
set -euo pipefail

# Count unique IP addresses from a log
grep -oE "([0-9]{1,3}\.){3}[0-9]{1,3}" /var/log/syslog 2>/dev/null \
    | sort \
    | uniq -c \
    | sort -rn \
    | head -10

# Find largest files in a directory
find /var/log -type f -printf "%s %p\n" 2>/dev/null \
    | sort -rn \
    | head -5 \
    | while read -r size path; do
        printf "%10d  %s\n" "$size" "$path"
    done

# Pipeline with tee for branching output
seq 1 100 \
    | tee >(grep -c "even" > /dev/stderr) \
    | grep -v "0" \
    | wc -l

# Named pipes for inter-process communication
fifo="/tmp/myfifo-$$"
mkfifo "$fifo"

( seq 1 10 > "$fifo" ) &
( while read -r num; do
    echo "Received: $num"
  done < "$fifo"
) &

wait
rm -f "$fifo"

# Complex pipeline for log analysis
cat /var/log/syslog 2>/dev/null \
    | grep -E "ERROR|WARN" \
    | awk "{print \$1, \$2, \$NF}" \
    | sort \
    | uniq -c \
    | sort -rn \
    | head -20 \
    | column -t