管道 - 过滤器（软考架构师架构风格）

管道 - 过滤器（Pipe-and-Filter）架构风格

一、核心定义与本质

管道过滤器是数据流驱动的经典架构模式，属于结构化 / 数据流架构，最早源于 Unix Shell 设计思想。

核心组成两部分

1. 过滤器（Filter）
   • 独立、无状态、单一职责的数据处理单元；
   • 只做一件事：接收输入数据流 → 转换 / 加工 / 过滤数据 → 输出数据流；
   • 不共享全局数据，不依赖其他过滤器内部状态，可独立复用、替换、并行运行。
2. 管道（Pipe）
   • 连接两个过滤器的通道，只负责传输数据流，不修改数据；
   • 标准单向传输：上游 Filter 输出 → Pipe → 下游 Filter 输入；
   • 解耦：过滤器之间互不感知对方实现，只约定数据格式。

核心思想

数据流经一系列独立处理组件，每个组件只完成单一处理，组件之间仅通过数据流通信。

二、关键特性

优点

1. 高复用：单个过滤器可在多个流程中重复使用（如日志过滤、格式转换）；
2. 易修改扩展：新增处理逻辑只需插入新过滤器，无需改动原有代码；
3. 可并行执行：多过滤器可流水线并发处理，提升吞吐量；
4. 松耦合：过滤器完全隔离，替换其中一个不影响整条链路；
5. 可调试：管道中间可截流、打印数据流，快速定位异常；
6. 无状态天然容错：单个过滤器故障不直接击穿全链路（可加熔断）。

缺点

1. 数据格式强约束：上下游必须统一数据流格式，格式变更改动全链路；
2. 额外 IO 开销：数据反复序列化 / 传输，大数据场景性能损耗明显；
3. 不适合交互场景：同步、低延迟人机交互不适合（适合批处理、流式处理）；
4. 复杂分支难维护：多分支、循环管道会大幅提升架构复杂度。

三、管道过滤器常见拓扑结构

1. 线性串联（最基础）Filter1 → Pipe → Filter2 → Pipe → Filter3
2. 分叉（广播）一个过滤器输出，通过管道分发给多个下游过滤器并行处理
3. 汇聚（合并）多个过滤器输出管道汇入同一个下游过滤器（如多日志合并）
4. 循环管道处理后数据回流上游再次过滤（较少用，易死循环）

四、最经典真实底层示例：Unix/Linux Shell 命令（原生管道过滤器）

原理

Unix 每个命令就是Filter，|符号就是Pipe； 每个命令默认：标准输入 stdin、标准输出 stdout，管道把前一个 stdout 接到后一个 stdin。

示例 1：日志过滤完整链路

# 原始日志文件读取(Filter1) → 筛选错误日志(Filter2) → 提取时间字段(Filter3) → 排序(Filter4) → 输出文件(Filter5) cat app.log | grep "ERROR" | awk '{print $1,$2}' | sort > error_time.log

拆解对应架构：

1. Filter1：cat— 读取文件，输出原始日志流
2. Pipe1：|— 传输全部日志文本
3. Filter2：grep "ERROR"— 过滤器：只保留包含 ERROR 的行
4. Pipe2：|— 传输错误日志行
5. Filter3：awk— 过滤器：截取每行前两列（日期、时间）
6. Pipe3：|— 传输时间文本流
7. Filter4：sort— 过滤器：按时间排序
8. Pipe4：>— 管道写入目标文件

特点：

• 每个命令独立、单一职责，可单独拿出来使用；
• 管道只传文本，不处理业务；
• 随意增删过滤器：加head -10只看前 10 条，不改动原有逻辑。

五、工程级真实业务系统示例（后端开发常用）

示例 1：实时日志处理系统（ELK 架构 = 标准管道过滤器）

整体链路：原始日志 → Filebeat → Logstash → Elasticsearch → Kibana 逐个映射：

1. Filter1 Filebeat：采集服务器日志文件、切割日志流
2. Pipe1：网络 TCP 通道传输日志原始文本
3. Filter2 Logstash（多过滤器串联）
   • grok 过滤器：解析非结构化日志为 JSON 结构化数据
   • date 过滤器：统一时间字段格式
   • drop 过滤器：丢弃无用调试日志
   • mutate 过滤器：新增业务标签字段
4. Pipe2：http 管道传输结构化 JSON
5. Filter3 Elasticsearch：存储、索引、分词过滤数据
6. Pipe3：查询数据流
7. Filter4 Kibana：聚合、图表过滤、可视化输出

特点：任意一层过滤器可替换（Filebeat 替换为 Flink 采集，Logstash 替换为 Flink Transform），完全符合管道过滤器设计。

示例 2：大数据实时计算 Flink/Spark Streaming

流式计算是管道过滤器工业级落地： 数据流链路：消息队列 (Kafka) → 数据源 Filter → 清洗 Filter → 转换 Filter → 聚合 Filter → 入库 Filter

• 每个算子 (map/filter/flatMap) = 独立过滤器；
• 上下游算子之间的数据通道 = 管道；
• 可随意插入新算子（脱敏、风控过滤），不修改原有计算逻辑。

业务场景：电商用户行为流处理

1. Filter1：读取 Kafka 用户点击日志
2. Filter2：清洗：过滤非法埋点、空数据
3. Filter3：脱敏：手机号、身份证加密处理
4. Filter4：转换：拆分行为字段、关联商品基础数据
5. Filter5：聚合：统计每小时商品点击量
6. Filter6：输出：写入 MySQL 报表库

六、日志处理Demo

package test; import java.util.ArrayList; import java.util.List; import java.util.Random; // ===================== 统一数据流载体（管道传输的数据） ===================== class DataRecord { private String content; public DataRecord(String content) { this.content = content; } public String getContent() { return content; } } // ===================== 过滤器顶层接口 所有处理单元实现该接口 ===================== interface Filter { List<DataRecord> process(List<DataRecord> input); } // ===================== 管道工具类：串联所有过滤器 ===================== class PipeLine { // 链式执行所有过滤器，数据在管道中流转 public static List<DataRecord> run(List<DataRecord> sourceData, Filter... filters) { List<DataRecord> data = sourceData; for (Filter filter : filters) { // 管道传输数据到下一个过滤器 data = filter.process(data); } return data; } } // ===================== 自定义实体：用户行为 ===================== class UserAction { private String userId; private String action; private Long productId; public UserAction(String userId, String action, Long productId) { this.userId = userId; this.action = action; this.productId = productId; } @Override public String toString() { return "UserAction{" + "userId='" + userId + '\'' + ", action='" + action + '\'' + ", productId=" + productId + '}'; } } // ===================== 各个独立过滤器（对应Flink算子） ===================== // 过滤器1：模拟数据源，批量生成原始日志 class SourceFilter implements Filter { @Override public List<DataRecord> process(List<DataRecord> input) { List<DataRecord> result = new ArrayList<>(); Random random = new Random(); String[] actions = {"click", "buy", "collect", "invalid"}; // 模拟生成10条埋点日志 for (int i = 0; i < 10; i++) { String uid = "u" + random.nextInt(9999); String act = actions[random.nextInt(actions.length)]; long pid = random.nextInt(1000); String log = uid + "," + act + "," + pid; result.add(new DataRecord(log)); } return result; } } // 过滤器2：清洗过滤，丢弃非法脏数据 class CleanFilter implements Filter { @Override public List<DataRecord> process(List<DataRecord> input) { List<DataRecord> result = new ArrayList<>(); for (DataRecord record : input) { String line = record.getContent(); String[] arr = line.split(","); // 过滤格式错误、非法行为 if (arr.length != 3) { continue; } String action = arr[1]; if ("click".equals(action) || "buy".equals(action) || "collect".equals(action)) { result.add(record); } } return result; } } // 过滤器3：字符串日志转结构化对象（模拟map转换算子） class ConvertFilter implements Filter { // 转换后存储对象，单独输出 public List<UserAction> convert(List<DataRecord> input) { List<UserAction> list = new ArrayList<>(); for (DataRecord record : input) { String[] arr = record.getContent().split(","); UserAction action = new UserAction(arr[0], arr[1], Long.valueOf(arr[2])); list.add(action); } return list; } @Override public List<DataRecord> process(List<DataRecord> input) { return input; } } // 过滤器4：用户ID脱敏处理 class MaskFilter { public List<UserAction> mask(List<UserAction> input) { List<UserAction> res = new ArrayList<>(); for (UserAction ua : input) { String rawId = ua.toString().split("userId='")[1].split("'")[0]; String maskId = rawId.substring(0, 2) + "****"; String action = ua.toString().split("action='")[1].split("'")[0]; String pid = ua.toString().split("productId=")[1].split("}")[0]; res.add(new UserAction(maskId, action, Long.valueOf(pid))); } return res; } } // ===================== 主程序：组装整条管道过滤器链路 ===================== public class SimplePipeFilterDemo { public static void main(String[] args) { // 1. 组装过滤器链路 Filter source = new SourceFilter(); Filter clean = new CleanFilter(); ConvertFilter convert = new ConvertFilter(); MaskFilter mask = new MaskFilter(); // 2. 管道流转数据：生成原始数据 -> 清洗 List<DataRecord> rawData = source.process(new ArrayList<>()); List<DataRecord> cleanData = PipeLine.run(rawData, clean); // 3. 管道下游：转换实体 + 脱敏 List<UserAction> entityList = convert.convert(cleanData); List<UserAction> finalResult = mask.mask(entityList); // 4. 输出过滤器：打印最终结果 System.out.println("===== 管道过滤器最终输出结果 ====="); for (UserAction item : finalResult) { System.out.println(item); } } }