java8-stream 4 聊聊串行与并行
2018-12-12 本文已影响6人
MikeLi666
在进行接下来的话题前，我们来谈谈并行，串行的问题。
我们本节要弄明白的问题是，串行和并行，执行的流程是什么样的。
不同的中间操作（运功）会对流程有什么样的影响。
说真心的我真的不想大段大段贴代码。但是为了让各位看官能明白我在做什么，我实在是没有其他什么特别好的办法了。
串行 一个线程做完所有事情
优点:不存在什么线程安全问题，保证处理的先后顺序。
缺点:速度比较慢。
串行 演示.gif
来看一个简单的例子，我需要将集合中的1，2，3，4，5。每个元素成先乘以10，然后再乘以10。将结果写到一个list。
当然进行这个例子，不是无聊到每个数字乘以100。而是要向大家演示串行时，stream是如何工作的。
public class Test20181211 {
@Test
public void test1() {
List<Integer> list = Arrays.asList(1, 2, 3, 4, 5);
List<Integer> collect = list.stream()
.map(this::map1)
.map(this::map2)
.collect(Collectors.toList());
System.out.println(collect);
}
private Integer map1(Integer i) {
int r = i * 10;
System.out.println("线程:" + Thread.currentThread().getName() + "\t方法 map1\t" + "收到:" + i + "\t" + "输出:" + r);
return r;
}
private Integer map2(Integer i) {
int r = i * 10;
System.out.println("线程:" + Thread.currentThread().getName() + "\t方法 map2\t" + "收到:" + i + "\t" + "输出:" + r);
return r;
}
}
线程:main 方法 map1 收到:1 输出:10
线程:main 方法 map2 收到:10 输出:100
线程:main 方法 map1 收到:2 输出:20
线程:main 方法 map2 收到:20 输出:200
线程:main 方法 map1 收到:3 输出:30
线程:main 方法 map2 收到:30 输出:300
线程:main 方法 map1 收到:4 输出:40
线程:main 方法 map2 收到:40 输出:400
线程:main 方法 map1 收到:5 输出:50
线程:main 方法 map2 收到:50 输出:500
[100, 200, 300, 400, 500]
不管运行几次，我们都会得到相同的结果。
从结果来看，main线程的处理过程是每次拿一个元素，然后顺序的进行一遍流程，做完以后进行下一个元素。最后将结果收集起来。
提问：如果是如下的处理过程。stream的处理过程是A还是B
.map(this::map1)
.map(this::map2)
.sorted((o1, o2) -> o2-o1)
.map(this::map3)
.map(this::map4)
A 获得元素->map1->map2->排序->map3->map4->获得元素->map1->map2->排序->map3->map4->循环到无元素->收集
B 获得元素->map1->map2->获得元素->map1->map2->循环到无元素->排序->获得元素->map3->map4->获得元素->map3->map4->循环到无元素->收集
答案是B
无需等待上游类型 ##### filter | map | flatMap | peek | peek |
需要等待上游类型 ##### distinct | limit | sorted
我们将中间操作分为两种类别。
需要等待上游类型的集中操作，都是需要上游流程处理结束以后，根据结果才能做出操作的。
public class Test20181211 {
@Test
public void test1() {
List<Integer> list = Arrays.asList(1, 2, 3, 4, 5);
List<Integer> collect = list.stream()
.map(this::map1)
.map(this::map2)
.sorted((o1, o2) -> o2 - o1)
.map(this::map3)
.map(this::map4)
.sorted()
.limit(2)
.collect(Collectors.toList());
System.out.println(collect);
}
private Integer map1(Integer i) {
int r = i * 10;
System.out.println("线程:" + Thread.currentThread().getName() + "\t方法 map1\t" + "收到:" + i + "\t" + "输出:" + r);
return r;
}
private Integer map2(Integer i) {
int r = i * 10;
System.out.println("线程:" + Thread.currentThread().getName() + "\t方法 map2\t" + "收到:" + i + "\t" + "输出:" + r);
return r;
}
private Integer map3(Integer i) {
int r = i * 10;
System.out.println("线程:" + Thread.currentThread().getName() + "\t方法 map3\t" + "收到:" + i + "\t" + "输出:" + r);
return r;
}
private Integer map4(Integer i) {
int r = i * 10;
System.out.println("线程:" + Thread.currentThread().getName() + "\t方法 map4\t" + "收到:" + i + "\t" + "输出:" + r);
return r;
}
}
线程:main 方法 map1 收到:1 输出:10
线程:main 方法 map2 收到:10 输出:100
线程:main 方法 map1 收到:2 输出:20
线程:main 方法 map2 收到:20 输出:200
线程:main 方法 map1 收到:3 输出:30
线程:main 方法 map2 收到:30 输出:300
线程:main 方法 map1 收到:4 输出:40
线程:main 方法 map2 收到:40 输出:400
线程:main 方法 map1 收到:5 输出:50
线程:main 方法 map2 收到:50 输出:500
这里做了逆序排序
线程:main 方法 map3 收到:500 输出:5000
线程:main 方法 map4 收到:5000 输出:50000
线程:main 方法 map3 收到:400 输出:4000
线程:main 方法 map4 收到:4000 输出:40000
线程:main 方法 map3 收到:300 输出:3000
线程:main 方法 map4 收到:3000 输出:30000
线程:main 方法 map3 收到:200 输出:2000
线程:main 方法 map4 收到:2000 输出:20000
线程:main 方法 map3 收到:100 输出:1000
线程:main 方法 map4 收到:1000 输出:10000
这里做了正序排序 还有限制
[10000, 20000]
所以最终应该是这个样子
串行 综合演示.gif
并行 多个线程一起做
优点:处理速度快
缺点:线程安全处理不得当会引发灾难。
并行 演示.gif
public class Test20181211 {
@Test
public void test1() {
List<Integer> list = Arrays.asList(1, 2, 3, 4, 5);
//注意 注意 注意 我只改动了这里 生成了一个并行流
List<Integer> collect = list.parallelStream()
.map(this::map1)
.map(this::map2)
.sorted((o1, o2) -> o2 - o1)
.map(this::map3)
.map(this::map4)
.sorted()
.limit(2)
.collect(Collectors.toList());
System.out.println(collect);
}
private Integer map1(Integer i) {
int r = i * 10;
System.out.println("线程:" + Thread.currentThread().getName() + "\t方法 map1\t" + "收到:" + i + "\t" + "输出:" + r);
return r;
}
private Integer map2(Integer i) {
int r = i * 10;
System.out.println("线程:" + Thread.currentThread().getName() + "\t方法 map2\t" + "收到:" + i + "\t" + "输出:" + r);
return r;
}
private Integer map3(Integer i) {
int r = i * 10;
System.out.println("线程:" + Thread.currentThread().getName() + "\t方法 map3\t" + "收到:" + i + "\t" + "输出:" + r);
return r;
}
private Integer map4(Integer i) {
int r = i * 10;
System.out.println("线程:" + Thread.currentThread().getName() + "\t方法 map4\t" + "收到:" + i + "\t" + "输出:" + r);
return r;
}
}
每次输出的结果，就没有那么固定了。有的时候线程1先做完了工作，有的时候线程2先做完了工作。
线程:main 方法 map1 收到:5 输出:50
线程:ForkJoinPool.commonPool-worker-4 方法 map1 收到:4 输出:40
线程:ForkJoinPool.commonPool-worker-1 方法 map1 收到:2 输出:20
线程:ForkJoinPool.commonPool-worker-3 方法 map1 收到:1 输出:10
线程:ForkJoinPool.commonPool-worker-3 方法 map2 收到:10 输出:100
线程:ForkJoinPool.commonPool-worker-2 方法 map1 收到:3 输出:30
线程:ForkJoinPool.commonPool-worker-1 方法 map2 收到:20 输出:200
线程:ForkJoinPool.commonPool-worker-4 方法 map2 收到:40 输出:400
线程:main 方法 map2 收到:50 输出:500
线程:ForkJoinPool.commonPool-worker-2 方法 map2 收到:30 输出:300
在这个位置 sorted 大家等排序完成以后继续抢工作，排序工作开始的前提是上游stream里面的内容都处理结束了。
线程:main 方法 map3 收到:100 输出:1000
线程:ForkJoinPool.commonPool-worker-3 方法 map3 收到:200 输出:2000
线程:ForkJoinPool.commonPool-worker-3 方法 map4 收到:2000 输出:20000
线程:ForkJoinPool.commonPool-worker-2 方法 map3 收到:400 输出:4000
线程:ForkJoinPool.commonPool-worker-2 方法 map4 收到:4000 输出:40000
线程:ForkJoinPool.commonPool-worker-1 方法 map3 收到:500 输出:5000
线程:ForkJoinPool.commonPool-worker-4 方法 map3 收到:300 输出:3000
线程:ForkJoinPool.commonPool-worker-1 方法 map4 收到:5000 输出:50000
线程:main 方法 map4 收到:1000 输出:10000
线程:ForkJoinPool.commonPool-worker-4 方法 map4 收到:3000 输出:30000
在这个位置 sorted 大家等排序完成以后继续抢工作，排序工作开始的前提是上游stream里面的内容都处理结束了。
排序结束后，limit前两个，输出到下list
[10000, 20000]
我们可以看到，我们自己在没有创建线程的情况下，仅仅通过parallelStream 获取并行流。就开启了并行处理能力。用起来十分的方便。通过线程名字，我们也能发现
是ForkJoinPool线程池中的
不同commonPool-worker-*来帮我们做了并行的工作。
下一节中 我们会讲讲，错误使用并行流造成的影响。
