java8-stream 4 聊聊串行与并行

在进行接下来的话题前，我们来谈谈并行，串行的问题。

我们本节要弄明白的问题是，串行和并行，执行的流程是什么样的。
不同的中间操作（运功）会对流程有什么样的影响。

说真心的我真的不想大段大段贴代码。但是为了让各位看官能明白我在做什么，我实在是没有其他什么特别好的办法了。

串行 一个线程做完所有事情
优点:不存在什么线程安全问题，保证处理的先后顺序。
缺点:速度比较慢。

串行 演示.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-*来帮我们做了并行的工作。

下一节中 我们会讲讲，错误使用并行流造成的影响。