管道模式 流处理

（一）介绍

管道这个名字源于自来水厂的原水处理过程。原水要经过管道，一层层地过滤、沉淀、去杂质、消毒，到管道另一端形成纯净水。我们不应该把所有原水的过滤都放在一个管道中去提纯，而应该把处理过程进行划分，把不同的处理分配在不同的阀门上，第一道阀门调节什么，第二道调节什么……最后组合起来形成过滤纯净水的管道。
这种处理方式实际上体现了一种分治（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);

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

（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);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

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);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

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);
    }
}
1
2
3
4
5
6
7
8

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

public class SecondValve extends ValveBase {
    @Override
    public void invoke(String s) {
        s = s.replace("22", "second");
        System.out.println("SecondValve阀门处理后结果" + s);
        getNext().invoke(s);
    }
}
1
2
3
4
5
6
7
8

（3）尾阀门：

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

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();
            }
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

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);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

6．执行客户端程序并输出结果

【完整代码】
Github代码