可组合性的三个要素
├── 统一接口 —— 所有组件使用相同的输入/输出方式
├── 独立性   —— 每个组件不依赖其他组件的内部实现
└── 无状态   —— 组件不保留上次调用的状态（管道中的工具）

# Unix 提供三种主要的组合机制

# 1. 管道（Pipe）—— 运行时组合
cmd1 | cmd2 | cmd3

# 2. Shell 脚本 —— 编排组合
#!/bin/bash
result=$(cmd1 | cmd2)
cmd3 "$result"

# 3. 文件系统 —— 持久化组合
cmd1 < input.txt > temp.txt
cmd2 < temp.txt > output.txt
rm temp.txt

┌──────────┐     ┌──────────┐     ┌──────────┐
│  输入源  │ ──→ │  过滤器  │ ──→ │  输出目标 │
│  (file)  │     │  (grep)  │     │  (file)  │
└──────────┘     └──────────┘     └──────────┘
     ↑                                    │
     │         可以链式连接                 ↓
     │    ┌──────────┐     ┌──────────┐    │
     └──  │  过滤器  │ ──→ │  过滤器  │ ──┘
          │  (sort)  │     │  (uniq)  │
          └──────────┘     └──────────┘

# 需求：从 Nginx 日志中找出访问量最高的 IP，
# 排除已知的健康检查 IP，生成 Top 20 报告

cat /var/log/nginx/access.log |     # 1. 读取日志
    grep -v "health_check" |         # 2. 排除健康检查
    awk '{print $1}' |               # 3. 提取 IP
    sort |                            # 4. 排序（uniq 需要）
    uniq -c |                         # 5. 统计次数
    sort -rn |                        # 6. 按次数降序
    head -20 |                        # 7. 取 Top 20
    awk '{printf "%5d  %s\n", $1, $2}' # 8. 格式化输出

管道的接口契约
├── 输入格式：行文本流（每行以 \n 结尾）
├── 输出格式：行文本流（每行以 \n 结尾）
├── 编码：与 locale 设置一致（通常是 UTF-8）
├── 退出码：0=成功，非0=失败
└── 错误处理：错误信息输出到 stderr

# ❌ 1. 输出包含颜色代码（破坏下游解析）
ls --color=always | grep "file"    # grep 可能匹配到 ANSI 颜色码

# ✅ 正确做法
ls --color=auto | grep "file"      # auto 模式：管道中不输出颜色

# ❌ 2. 输出包含进度信息（破坏数据流）
wget -O - url 2>/dev/null | grep "pattern"
# 如果 wget 向 stdout 输出进度信息，会污染数据

# ✅ 正确做法
wget -q -O - url | grep "pattern"  # -q 静默模式

# ❌ 3. 输出二进制数据到文本管道
cat image.png | grep "text"        # grep 搜索二进制数据会产生乱码

# ✅ 正确做法
strings image.png | grep "text"    # 只提取可打印字符串

#!/bin/bash
# 一个设计良好的可组合工具示例

# 1. 支持 stdin 和文件参数
input_source() {
    if [ $# -eq 0 ]; then
        cat -  # 从 stdin 读取
    else
        cat "$@"  # 从文件读取
    fi
}

# 2. 正常输出到 stdout，错误输出到 stderr
process() {
    input_source "$@" | while IFS= read -r line; do
        # 处理逻辑
        result=$(echo "$line" | some_operation)
        if [ $? -eq 0 ]; then
            echo "$result"         # 正常结果 → stdout
        else
            echo "Error: $line" >&2  # 错误信息 → stderr
        fi
    done
}

# 3. 用退出码表示结果
process "$@"
exit $?

#!/usr/bin/env python3
"""
可组合的命令行工具：统计文本中的词频
支持 stdin 和文件参数，输出到 stdout
"""
import sys
from collections import Counter

def read_input(args):
    """统一的输入接口"""
    if not args:
        yield from sys.stdin
    else:
        for filename in args:
            with open(filename) as f:
                yield from f

def word_freq(args):
    """主逻辑"""
    counter = Counter()
    for line in read_input(args):
        words = line.strip().lower().split()
        counter.update(words)
    
    # 输出到 stdout（可组合）
    for word, count in counter.most_common():
        print(f"{count}\t{word}")

if __name__ == "__main__":
    word_freq(sys.argv[1:])

# 使用示例
cat article.txt | python3 wordfreq.py | head -20
python3 wordfreq.py *.txt | sort -rn | head -20

# 多个日志文件汇聚分析
cat /var/log/app/*.log | grep "ERROR" | sort | uniq -c | sort -rn

# 使用 find + xargs 实现扇入
find /var/log -name "*.log" -print0 | xargs -0 cat | grep "ERROR"

# 使用进程替换
cat <(cat access.log) <(cat error.log) | grep "pattern"

# 多个远程日志汇聚
ssh server1 "cat /var/log/app.log" > /tmp/s1.log
ssh server2 "cat /var/log/app.log" > /tmp/s2.log
cat /tmp/s1.log /tmp/s2.log | grep "ERROR" | sort | uniq -c

# 使用 tee 实现扇出
cat data.txt | tee >(grep "error" > errors.txt) | \
                   tee >(grep "warn" > warnings.txt) | \
                   grep "info" > info.txt

# 使用 tee 写入多个文件
echo "broadcast message" | tee file1.txt file2.txt file3.txt > /dev/null

# 使用命名管道实现扇出
mkfifo /tmp/pipe1 /tmp/pipe2
cat data.txt | tee /tmp/pipe1 > /tmp/pipe2 &
consumer1 < /tmp/pipe1 &
consumer2 < /tmp/pipe2 &
wait

# ETL（Extract-Transform-Load）模式
#!/bin/bash

# Stage 1: Extract
extract() {
    curl -s "https://api.example.com/data" | jq '.results[]'
}

# Stage 2: Transform
transform() {
    jq '{name: .name, value: .value * 1.1}'
}

# Stage 3: Load
load() {
    while IFS= read -r record; do
        echo "$record" | jq -r '[.name, .value] | @csv'
    done >> /data/output.csv
}

# 组合
extract | transform | load

#!/bin/bash
# 根据条件路由到不同的处理分支

cat access.log | while IFS= read -r line; do
    status=$(echo "$line" | awk '{print $9}')
    case "$status" in
        2*) echo "$line" >> success.log ;;
        3*) echo "$line" >> redirect.log ;;
        4*) echo "$line" >> client_error.log ;;
        5*) echo "$line" >> server_error.log ;;
    esac
done

# 合并多个已排序的文件
sort -m sorted1.txt sorted2.txt sorted3.txt

# 合并远程排序结果
sort -m <(ssh s1 "sort /data/part1.txt") \
        <(ssh s2 "sort /data/part2.txt") \
        <(ssh s3 "sort /data/part3.txt")

# 安装
# apt install parallel / brew install parallel

# 基本用法：并行执行命令
cat urls.txt | parallel curl -s {}

# 并行处理文件
find . -name "*.jpg" | parallel -j 4 convert {} -resize 50% thumb_{/}

# 控制并行度
seq 1 100 | parallel -j 10 "echo processing {}"

# 带进度条
find . -name "*.log" | parallel --progress "gzip {}"

# 远程并行执行
parallel -S server1,server2 "echo {} on {/}" ::: *.txt

# xargs 也支持简单的并行
find . -name "*.jpg" | xargs -P 4 -I{} convert {} -resize 50% thumb_{}

# -P 4 表示同时运行 4 个进程
# -I{} 指定替换标记

管道的性能特点
├── 每个 | 创建两个进程（写入端和读取端）
├── 数据在内核缓冲区中复制多次
├── 管道缓冲区大小限制（通常 64KB-1MB）
├── 长管道链的进程管理开销
└── 文本解析的 CPU 开销

# 对于大规模数据，考虑使用更高效的工具
# ❌ 低效：多次遍历
cat bigfile.txt | grep "pattern" | awk '{print $1}' | sort | uniq

# ✅ 高效：单次遍历
awk '/pattern/ { count[$1]++ } END { for (k in count) print count[k], k }' bigfile.txt | sort -rn

# ❌ 低效：管道中的多次 IO
cat bigfile.txt | sort | uniq | wc -l

# ✅ 高效：使用 sort -u
sort -u bigfile.txt | wc -l

# 管道链过长时，调试很困难
# ❌ 难以调试的长管道
cat data.txt | tr ',' '\t' | cut -f2 | grep -v "^$" | sort | uniq -c | sort -rn | head -10

# ✅ 方法 1: 在管道中间插入 tee 查看中间结果
cat data.txt | tr ',' '\t' | tee /dev/stderr | cut -f2 | grep -v "^$" | sort | uniq -c | sort -rn | head -10

# ✅ 方法 2: 拆分为多步
cat data.txt | tr ',' '\t' > /tmp/step1
cut -f2 /tmp/step1 | grep -v "^$" | sort > /tmp/step2
uniq -c /tmp/step2 | sort -rn | head -10

# ✅ 方法 3: 使用 set -x 逐行调试
#!/bin/bash
set -x
cat data.txt | tr ',' '\t' | cut -f2 | grep -v "^$" | sort | uniq -c | sort -rn | head -10

# Docker 镜像构建也遵循 Unix 组合哲学
# 每个 RUN 指令是一个"过滤器"，通过层（layer）组合

# 多阶段构建
FROM golang:1.22 AS builder
RUN go build -o app .

FROM alpine:3.19
COPY --from=builder /app /app
CMD ["/app"]

# Docker 管道
docker logs container_name 2>&1 | grep "error" | sort | uniq -c

# Kubernetes 的声明式 API 也是组合性的体现
# 每个资源（Deployment, Service, ConfigMap）是独立组件
# 通过 label selector 组合

apiVersion: apps/v1
kind: Deployment
metadata:
  name: web
spec:
  selector:
    matchLabels:
      app: web
  template:
    metadata:
      labels:
        app: web
    spec:
      containers:
      - name: web
        image: nginx
---
apiVersion: v1
kind: Service
metadata:
  name: web-service
spec:
  selector:
    app: web  # 通过 label 组合
  ports:
  - port: 80

Unix 管道 → 微服务架构
├── cat → 数据源（数据库、API）
├── grep → 过滤服务
├── awk → 转换服务
├── sort → 排序服务
├── uniq → 聚合服务
└── head → 限流服务

每个微服务：
├── 做一件事做好它
├── 通过标准化接口（HTTP/gRPC）通信
├── 可以独立部署和替换
└── 通过编排工具（Kubernetes）组合