Java多线程模式之流水线模式

流水线

什么是流水线？

在计算机中，对于一条具体的指令执行过程，通常可以分为五个部分：取指令，指令译码，取操作数，运算（ALU），写结果。前三步由指令控制器完成，后两步则由运算器完成。按照传统的方式，所有指令顺序执行，那么先是指令控制器工作，完成第一条指令的前三步，然后运算器工作，完成后两步，第一条指令执行完毕。然后第二条指令又是先指令控制器工作，完成前三步，然后运算器，完成第二条指令的后两部…… 传统方式有个很大的缺点就是，指令只能一条一条地执行，仔细分析一下就会发现，这种方式存在很大的资源浪费：即同一时刻，要么指令控制器工作，运算器闲着；要么运算器工作，指令控制器闲着。这样一方面资源得不到有效利用，另一方面就是工作效率很低。流水线的出现就是为了解决这个问题，下面我们来看一下流水线的工作模式：假设有两个指令INS_A和INS_B,它们的执行分别要经过A,B,C,D四个过程，假设A到D四个过程分别由四个硬件元件完成。按照传统的方式，它们的流程如下：传统方式这种方式的缺点就是，只能一条指令一条指令的执行，并且当指令执行到过程B的时候，处理过程A和CD的元件是处于空闲状态的。

流水线方式如下：

说明一下，通过流水线的方式，当INS_A指令执行完过程A之后，处理过程A的元件就空闲了，此时我们就开始处理指令INS_B的A阶段，这样一来，INS_B指令只需要等到INS_A的A过程执行完成就可以继续执行了，这样以来就在很大程度上提高了效率。

流水线中断

使用流水线能够很大程度提高程序执行效率，这点是毋庸置疑的，但是，在系统中，每当引入一个新的模式或者组件的时候，我们就需要对应处理该模式或者组件所带来的问题，那么引入流水线的一个很大的问题就是流水线中断。

产生中断一般是由两个原因造成的，一个是“相关”，一个是“条件转移”。相关:指的是一条指令的执行需要依赖前一条指令的执行结果。拿上面的例子，假设INS_B指令在执行过程C的时候，需要使用INS_A指令过程D的结果，那么指令INS_B在执行到C的时候，由于A指令的D过程还没有执行完成，所以此时INS_B就不能继续执行，否则就会拿到错误结果。所以此时就需要将流水线中断，从而等待指令INS_A的D过程的结束。如下图所示：

条件转移:如果一条指令是条件转移指令，即指令执行结果是根据条件发生变化的，那么系统就不清楚下面应该执行哪一条指令，这时就必须等第一条指令的判断结果出来才能执行第二条指令。条件转移所造成的流水线停顿甚至比相关还要严重的多。

所以：流水线虽然能够提高整体运行效率，但是在某些情况下，需要中断流水线，以保证程序运行正确，而流水线的中断又会降低系统运行效率。

流水线在Java中的应用

目前在java提供的实现中，并没有用到流水线模式，但是我们知道JVM虚拟机在运行时对我们的代码进行的指令重排序，目的就是为了减少流水线的中断，从而提高流水线运行效率。

在实际的项目开发中，我们也可以学习流水线模式的思想，将业务流程拆分成多个子流程，然后采用流水线的方式进行，以减少程序等待。比如有一个操作涉及到1）查询数据库，2）本地处理数据，3）远程RPC通知结果三个过程，其中过程1和过程3都涉及到网络IO操作，所以整体运行是比较耗时的。但是我们可以采用流水线作业的方式，这样就可以充分利用三部分的资源。提高系统的整体运行效率。

Commons Pipeline

apache基金会下的一个项目，提供了流水线操作的框架。参考这里 PS:这个框架貌似好久都没有维护过了，不知道为啥。。

自己实现一个简单的流水线

在黄文海先生的《Java多线程实战指南(设计模式篇)》书中，详细讲解了流水线的java实现，另外，可以参考它书中的源码,点击这里

我也简单写了一个，假设有一个任务需要分为三步执行，那么可以将每一步抽象成一个任务阶段（TaskStage)，然后每个任务阶段都被提交到队列中，然后三个线程分别处理三个队列中的数据。 TaskStage

1package pipeline;
2
3/**
4 * 任务阶段，比如一个任务分为三个阶段，每一个阶段都有一个TaskStage与之对应
5 */
6public interface TaskStage {
7
8	/**
9	 * 处理任务，返回值为该任务阶段的下一个阶段
10	 */
11	public TaskStage process();
12}
13

三个具体的任务阶段：数据库查询任务阶段

1package pipeline;
2
3/**
4 * 查询数据库阶段
5 */
6public class DBQueryTaskStage implements TaskStage {
7
8	@Override
9	public TaskStage process() {
10		String result = queryDB();
11		System.out.println(result);
12		return new CalculateTaskStage(result);
13	}
14
15	/**
16	 * 模拟查询数据库过程
17	 * 
18	 * @return
19	 */
20	private String queryDB() {
21		try {
22			Thread.sleep(1000 * 20);
23		} catch (InterruptedException e) {
24			e.printStackTrace();
25		}
26		return "DBQueryTaskStage";
27	}
28
29}

计算处理结果任务阶段

1package pipeline;
2
3public class CalculateTaskStage implements TaskStage {
4	private String data;
5
6	public CalculateTaskStage(String data) {
7		super();
8		this.data = data;
9	}
10
11	@Override
12	public TaskStage process() {
13		try {
14			Thread.sleep(1000 * 5);
15		} catch (InterruptedException e) {
16			e.printStackTrace();
17		}
18		String result = calculate(data);
19		System.out.println(result);
20		return new RpcTaskStage(result);
21	}
22
23	private String calculate(String data2) {
24		return data2 + ",CaculateTaskStage";
25	}
26
27}
28

远程RPC调用任务阶段

1package pipeline;
2
3public class RpcTaskStage implements TaskStage {
4
5	private String data;
6
7	public RpcTaskStage(String data) {
8		super();
9		this.data = data;
10	}
11
12	@Override
13	public TaskStage process() {
14		try {
15			Thread.sleep(1000 * 3);
16		} catch (InterruptedException e) {
17			e.printStackTrace();
18		}
19		System.out.println(data + ",RpcTaskStage");
20		return null;
21	}
22
23}
24

接下来需要有一个流水线类，用于任务调度

1package pipeline;
2
3import java.util.concurrent.BlockingDeque;
4import java.util.concurrent.LinkedBlockingDeque;
5import java.util.concurrent.TimeUnit;
6
7/**
8 * 整条流水线
9 */
10public class Pipeline {
11
12	// 用于存放任务的第一阶段
13	private BlockingDeque<TaskStage> firstStageQueue = new LinkedBlockingDeque<TaskStage>();
14
15	// 用于存储任务的第二阶段
16	private BlockingDeque<TaskStage> secondStageQueue = new LinkedBlockingDeque<TaskStage>();
17
18	// 用于存储任务的第三阶段
19	private BlockingDeque<TaskStage> thirdStageQueue = new LinkedBlockingDeque<TaskStage>();
20
21	public Pipeline() {
22		super();
23
24		// 启动三个线程，分别处理三个阶段的任务
25		new Thread() {
26
27			@Override
28			public void run() {
29				while (true) {
30					TaskStage taskStage = null;
31					try {
32						taskStage = firstStageQueue.poll(80, TimeUnit.SECONDS);
33					} catch (InterruptedException e1) {
34						e1.printStackTrace();
35					}
36					if (taskStage != null) {
37						TaskStage nextStage = taskStage.process();
38						try {
39							secondStageQueue.put(nextStage);
40						} catch (InterruptedException e) {
41						}
42					} else { // 取出来的数据为空，则终止（这种情况可能会有问题，这里做简单演示）
43						break;
44					}
45				}
46			}
47		}.start();
48
49		new Thread() {
50
51			@Override
52			public void run() {
53				while (true) {
54					TaskStage taskStage = null;
55					try {
56						taskStage = secondStageQueue.poll(80, TimeUnit.SECONDS);
57					} catch (InterruptedException e1) {
58						e1.printStackTrace();
59					}
60					if (taskStage != null) {
61						TaskStage nextStage = taskStage.process();
62						try {
63							thirdStageQueue.put(nextStage);
64						} catch (InterruptedException e) {
65						}
66					} else {
67						break;
68					}
69				}
70			}
71		}.start();
72
73		new Thread() {
74
75			@Override
76			public void run() {
77				while (true) {
78					TaskStage taskStage = null;
79					try {
80						taskStage = thirdStageQueue.poll(80, TimeUnit.SECONDS);
81					} catch (InterruptedException e) {
82						e.printStackTrace();
83					}
84					if (taskStage != null) {
85						taskStage.process();
86					} else {
87						break;
88					}
89				}
90			}
91		}.start();
92	}
93
94	public void process(TaskStage firstStage) {
95		try {
96			firstStageQueue.put(firstStage);
97		} catch (InterruptedException e) {
98			e.printStackTrace();
99		}
100	}
101
102}
103

主程序入口:

1package pipeline;
2
3import java.util.concurrent.locks.LockSupport;
4
5public class Main {
6	public static void main(String[] args) {
7		Pipeline line = new Pipeline();
8
9		for (int i = 0; i < 10; i++) {
10			line.process(new DBQueryTaskStage());
11		}
12
13		LockSupport.park();
14	}
15}
16

PS:上面的程序只是简单演示，会有一些线程调度的问题。

参考资料

流水线
《Java多线程实战指南(设计模式篇)》黄文海