谨慎使用 java8 并行流

问题引出
最近接触到一个关于排序的问题：模拟奥运会统计对每个国家获取的奖牌数进行排序，排序规则：排序顺序依次为金牌，银牌，铜牌的数量，若金牌的数量相等则比较银牌，银牌的数量相等，则比较铜牌，若三种奖牌数量都一样，则根据国家的英文字符排序，最终输出排序后的国家名称：
其实也不是一个很复杂的问题；第一时间想到java8提供的流操作，直使用sorted方法直接排序，可以同时支持多个字段的排序。代码如下：

public static void  sortedCountry(List<CountryInfo> countryInfos) {
        countryInfos.parallelStream().sorted(Comparator
                .comparingInt(CountryInfo::getGold)
                .thenComparingInt(CountryInfo::getSilver)
                .thenComparingInt(CountryInfo::getBronze)
                .thenComparing((a, b) -> b.getCountry().compareTo(a.getCountry()))
                .reversed())
               .forEach(e -> System.out.print(e.getCountry() + ", "));
    }


    private static class CountryInfo {
        private String country;
        private int gold;
        private int silver;
        private int bronze;

        public CountryInfo(String country, int gold, int silver, int bronze) {
            this.country = country;
            this.gold = gold;
            this.silver = silver;
            this.bronze = bronze;
        }

        public String getCountry() {
            return country;
        }

        public int getGold() {
            return gold;
        }

        public int getSilver() {
            return silver;
        }

        public int getBronze() {
            return bronze;
        }
    }

代码中sortedCountry是核心方法，分析该段代码的话，看起来并没有什么问题。但是运行的时候出问题了，如下：

测试结果

预期结果应该是china，japan，uk，aust，us。
最后发现，代码改成这样就行了。

public static void  sortedCountry(List<CountryInfo> countryInfos) {
        List<CountryInfo> collect = countryInfos.parallelStream().sorted(Comparator
                .comparingInt(CountryInfo::getGold)
                .thenComparingInt(CountryInfo::getSilver)
                .thenComparingInt(CountryInfo::getBronze)
                .thenComparing((a, b) -> b.getCountry().compareTo(a.getCountry()))
                .reversed())
                .collect(Collectors.toList());
        for (CountryInfo countryInfo : collect) {
            System.out.print(countryInfo.getCountry() + ", ");
        }
    }

运行结果如下：

运行结果

两段代码对比一下，可以发现，唯一不一样的地方在于对最终集合遍历操作，第一段代码在流中直接执行遍历，第二段代码使用for循环，所以问题应该在并行流操作时出现的问题。parallelStream工作原理采用forkjoin的思想，也就是采用jdk内部的forkjoin框架，等于开启多个线程对集合操作，第二段代码运行时，由于使用到流操作的执行结果，即collect对象，在进行for循环时候会用到，会一直阻塞，直到流被执行完成，才会开始遍历，所以结果是正确的。

分析原因
为什么会产生这样的结果：《java8实战》中这样描述：

流操作：无状态和有状态
你已经看到了很多的流操作。乍一看流操作简直是灵丹妙药，而且只要在从集合生成流的
时候把Stream换成parallelStream就可以实现并行。
当然，对于许多应用来说确实是这样，就像前面的那些例子。你可以把一张菜单变成流，
用filter选出某一类的菜肴，然后对得到的流做map来对卡路里求和，最后reduce得到菜单
的总热量。这个流计算甚至可以并行进行。但这些操作的特性并不相同。它们需要操作的内部
状态还是有些问题的。
诸如map或filter等操作会从输入流中获取每一个元素，并在输出流中得到0或1个结果。
这些操作一般都是无状态的：它们没有内部状态（假设用户提供的Lambda或方法引用没有内
部可变状态）。
但诸如reduce、sum、max等操作需要内部状态来累积结果。在上面的情况下，内部状态
很小。在我们的例子里就是一个int或double。不管流中有多少元素要处理，内部状态都是有界的。
相反，诸如sort或distinct等操作一开始都和filter和map差不多——都是接受一个
流，再生成一个流（中间操作），但有一个关键的区别。从流中排序和删除重复项时都需要知
道先前的历史。例如，排序要求所有元素都放入缓冲区后才能给输出流加入一个项目，这一操
作的存储要求是无界的。要是流比较大或是无限的，就可能会有问题（把质数流倒序会做什么
呢？它应当返回最大的质数，但数学告诉我们它不存在）。我们把这些操作叫作有状态操作。

所以第一段代码输出错误结果的原因是foreach操作读取了半成品的结果。想要得到正确的结果，可以使用第二段代码，也可以parallelStream改为stream，让流串行执行也是可以的，如下：

    public static void  sortedCountry(List<CountryInfo> countryInfos) {
        countryInfos.stream().sorted(Comparator
                .comparingInt(CountryInfo::getGold)
                .thenComparingInt(CountryInfo::getSilver)
                .thenComparingInt(CountryInfo::getBronze)
                .thenComparing((a, b) -> b.getCountry().compareTo(a.getCountry()))
                .reversed())
                .forEach(e -> System.out.print(e.getCountry() + ", "));

    }

串行流执行结果

总结（个人理解）
书中对有状态和无状态流描述的比较抽象，其实可以结合当前的例子来理解：有状态的流，前边使用的sort操作，它排序元素的时候比如说A，B，C，最坏的情况比较三次，即A和B比较，B和C比较，A和C比较，可以得到最终的排序结果；无状态的流，比如说sum，如果要求三个元素的和，A，B，C，需要两次相加就可以得到最终结果。其他操作也可以采用这种方式分析。
下图是《java8实战》的总结：

中间操作和终端操作