管道模式流处理

article/2025/9/14 11:53:19

（一）介绍

管道这个名字源于自来水厂的原水处理过程。原水要经过管道，一层层地过滤、沉淀、去杂质、消毒，到管道另一端形成纯净水。我们不应该把所有原水的过滤都放在一个管道中去提纯，而应该把处理过程进行划分，把不同的处理分配在不同的阀门上，第一道阀门调节什么，第二道调节什么……最后组合起来形成过滤纯净水的管道。
这种处理方式实际上体现了一种分治（Divid and Conquer）思想，这是一种古老且非常有效的思想。接下来，我们来看管道模式的实际应用。

（二）链式管道

一个典型的管道模式，会涉及以下3个主要的角色。
（1）阀门：处理数据的节点。（2）管道：组织各个阀门。（3）客户端：构造管道并调用。
对应现实生活中的管道，我们一般使用一个单向链表数据结构作为来实现，如图所示。这也是链式管道区别于拦截器模式之处。其实在功能上，拦截器、管道、过滤器、责任链有类似之处，在实际工作中，我们可以根据具体情况灵活选用。
在这里插入图片描述

为了便于理解，我找了一个管道阀门的图，结合上图可以更加容易理解。

【1】管道和阀门的整体结构就类似于一个链表结构，其中阀门是链表中的节点
【2】管道中可以添加和删除阀门，阀门按照顺序依次进行处理
【3】阀门可以根据需要对管道输入进行过滤处理。

基于上面的分析，我们可以按照下面的步骤实现一个简单的链式管道。

（1）阀门接口

/*** 阀门接口** @author zhangyu* @date 2022/12/4**/
public interface Valve {/*** 获取下一个阀门节点*/Valve getNext();/*** 设置下一个阀门节点*/void setNext(Valve v);/*** 当前阀门处理逻辑*/void invoke(String s);}

（2）管道接口：

/*** 管道接口** @author zhangyu* @date 2022/12/4**/
public interface Pipeline {/*** 获取管道中的第一个阀门节点*/public Valve getHead();/*** 获取管道中的第一个尾部阀门节点*/public Valve getTail();/*** 设置管道中的第一个尾部阀门节点*/public void setTail(Valve v);/*** 为管道添加阀门节点*/public void addValve(Valve v);
}

2．创建阀门的基础实现

/*** 阀门的基础实现** @author zhangyu* @date 2022/12/4**/
public abstract class ValveBase implements Valve {public Valve next;public Valve getNext() {return next;}public void setNext(Valve v) {next = v;}public abstract void invoke(String s);
}

3．实现具体的阀门
普通阀门一：当前模拟实现的场景是将输入字符串中的11替换为first字符串

public class FirstValve extends ValveBase {public void invoke(String s) {s = s.replace("11", "first");System.out.println("FirstValve阀门处理后结果" + s);getNext().invoke(s);}
}

（2）普通阀门二：：当前模拟实现的场景是将输入字符串中的22替换为second字符串

public class SecondValve extends ValveBase {@Overridepublic void invoke(String s) {s = s.replace("22", "second");System.out.println("SecondValve阀门处理后结果" + s);getNext().invoke(s);}
}

（3）尾阀门：

public class TailValve extends ValveBase {public void invoke(String s) {s = s.replace("33", "third");System.out.println("TailValve阀门处理后结果" + s);}
}

4．实现具体的管道

public class StandardPipeline implements Pipeline {protected Valve head;protected Valve tail;public Valve getHead() {return head;}public Valve getTail() {return tail;}public void setTail(Valve v) {tail = v;}/*** 往链表依次添加节点*/public void addValve(Valve v) {if (head == null) {// 如果链表为空则当前节点为头结点head = v;v.setNext(tail);} else {// 将当前节点添加依次添加到队列Valve current = head;while (current != null) {// 当前节点放入队列尾部，但是需要在tail尾结点前面if (current.getNext() == tail) {current.setNext(v);v.setNext(tail);break;}current = current.getNext();}}}
}

5．组装管道，实现客户端调用

public class Client {public static void main(String[] args) {String s = "11,22,33";System.out.println("原始输入 : " + s);StandardPipeline pipeline = new StandardPipeline();TailValve tail = new TailValve();FirstValve first = new FirstValve();SecondValve second = new SecondValve();// 设置管道尾部节点pipeline.setTail(tail);// 依次将节点加入至管道链表中pipeline.addValve(first);pipeline.addValve(second);// 从管道头部开始处理pipeline.getHead().invoke(s);}
}