上一节学习了Java8中比较常用的内置collector的用法。接下来就来理解下collector的组成。

Collector定义

Collector接口包含了一系列方法，为实现具体的归约操作(即收集器)提供了范本。我们已经看过了Collector接口中实现的许多收集器，例如toList或groupingBy。这也意味着你可以为Collector接口提供自己的实现，从而自由创建自定义归约操作。

要开始使用Collector接口，我们先来看看toList的实现方法，这个在日常中使用最频繁的东西其实也简单。

Collector接口定义了5个函数

public interface Collector
    
      {    Supplier
      supplier();    BiConsumer
      
        accumulator();    BinaryOperator
      
      combiner();    Function
      
        finisher();    Set
       
         characteristics();}
       
      
    

1. T是流中要收集的对象的泛型
2. A是累加器的类型，累加器是在收集过程中用于累积部分结果的对象。
3. R是收集操作得到的对象(通常但不一定是集合)的类型。

对于toList, 我们收集的对象是T, 累加器是List, 最终收集的结果也是一个List，于是创建ToListCollector如下：

public class ToListCollector
    
      implements Collector
     
      , List
      
       >
      
     
    

理解Collector几个函数

建立新的结果容器 supplier方法

supplier方法必须返回一个结果为空的Supplier，也就是一个无参数函数，在调用时，它会创建一个空的累加器实例，供数据收集过程使用。就个人通俗的理解来说，这个方法定义你如何收集数据，之所以提炼出来就是为了让你可以传lambda表达式来指定收集器。对于toList, 我们直接返回一个空list就好。

@Overridepublic Supplier
    
     
      > supplier() {    return ArrayList::new;}
     
    

累加器执行累加的具体实现 accumulator方法

accumulator方法会返回执行归约操作的函数，该函数将返回void。当遍历到流中第n个元素时，这个函数就会执行。函数有两个参数，第一个参数是累计值，第二参数是第n个元素。累加值与元素n如何做运算就是accumulator做的事情了。比如toList, 累加值就是一个List，对于元素n，当然就是add。

@Overridepublic BiConsumer
    
     
      , T> accumulator() {    return List::add;}
     
    

对结果容器应用最终转换 finisher方法

当遍历完流之后，我们需要对结果做一个处理，返回一个我们想要的结果。这就是finisher方法所定义的事情。finisher方法必须返回在累积过程的最后要调用的一个函数，以便将累加器对象转换为整个集合操作的最终结果， 这个返回的函数在执行时，会有个参数，该参数就是累积值，会有一个返回值，返回值就是我们最终要返回的东西。对于toList, 我最后就只要拿到那个收集的List就好，所以直接返回List。

@Overridepublic Function
    
     
      , List
      
       > finisher() {    return (i) -> i;}
      
     
    

对于接收一个参数，返回一个value，我们可以想到Function函数，正如finisher()的返回值。对于这个返回参数本身的做法，Function有个静态方法

static 
    
      Function
     
       identity() {    return t -> t;}
     
    

可以用Function.identity()代替上述lambda表达式。

合并两个结果容器 combiner

上面看起来似乎已经可以工作了，这是针对顺序执行的情况。我们知道Stream天然支持并行，但并行却不是毫无代价的。想要并行首先就必然要把任务分段，然后才能并行执行，最后还要合并。虽然Stream底层对我们透明的执行了并行，但如何并行还是需要取决于我们自己。这就是combiner要做的事情。combiner方法会返回一个供归约操作使用的函数，它定义了对流的各个子部分并行处理时，各个字部分归约所得的累加器要如何合并。对于toList而言，Stream会把流自动的分成几个并行的部分，每个部分都执行上述的归约，汇集成一个List。当全部完成后再合并成一个List。

@Overridepublic BinaryOperator
    
     
      > combiner() {    return (list1, list2) -> {        list1.addAll(list2);        return list1;    };}
     
    

这样，就可以对流并行归约了。它会用到Java7引入的分支/合并框架和Spliterator抽象。大概如下所示，

1. 原始流会以递归方式拆分为子流，直到定义流是否进一步拆分的一个条件为非(如果分布式工作单位太小，并行计算往往比顺序计算要慢，而且要是生成的并行任务比处理器内核数多很多的话就毫无意义了)。
2. 现在，所有的子流都可以并行处理，即对每个子流应用顺序归约算法。
3. 最后，使用收集器combiner方法返回的函数，将所有的部分结果两两合并。这时，会把原始流每次拆分得到的子流对应的结果合并起来。

characteristics方法

最后一个方法characteristics会返回一个不可变的Characteristics集合，它定义了收集器的行为--尤其是关于流是否可以并行归约，以及可以使用哪些优化的提示。

Characteristics是一个包含三个项目的枚举：

1. UNORDERED--归约结果不受流中项目的遍历和累积顺序的影响
2. CONCURRENT--accumulator函数可以从多个线程同时调用，且该收集器可以并行归约流。如果收集器没有标为UNORDERED, 那它仅在用于无序数据源时才可以并行归约。
3. IDENTITY_FINISH--这表明完成器方法返回的函数是一个恒等函数，可以跳过。这种情况下，累加器对象将会直接用做归约过程的最终结果。这也意味着，将累加器A不加检查地转换为结果R是安全的。

我们迄今为止ToListCollector是IDENTITY_FINISH的，因为用来累积流中元素的List已经是我们要的最终结果，用不着进一步转换了，但它并不是UNORDERED，因为用在有序流上的时候，我们还是希望顺序能够保留在得到到List中。最后，他是CONCURRENT的，但我们刚才说过了，仅仅在背后的数据源无序时才会并行处理。

上面这段话说的有点绕口，大概是说像Set生成的stream是无序的，这时候toList就可以并行。而ArrayList这种队列一样的数据结构则生成有序的stream，不能并行。

使用

直接传给collect方法就好。

List
    
      rs = dishes            .stream()            .collect(new ToListCollector<>());
    

我们这样费尽心思去创建一个toListCollector，一个是为了熟悉Collector接口的用法，一个是方便重用。当再遇到这样的需求的时候就可以直接用这个自定义的函数了，所以才有toList()这个静态方法。否则，其实collect提供了重载函数可以直接定义这几个函数。比如，可以这样实现toList

List
    
      dishes = dishes                    .stream()                    .collect(                        ArrayList::new, //supplier                        List::add, //accumulator                        List::addAll //combiner                    );
    

这种方法虽然简单，但可读性较差，而且当再次遇到这个需求时还要重写一遍，复用性差。

关于性能

对于stream提供的几个收集器已经可以满足绝大部分开发需求了，reduce提供了各种自定义。但有时候还是需要自定义collector才能实现。文中举例还是质数枚举算法。之前我们通过遍历平方根之内的数字来求质数。这次提出要用得到的质数减少取模运算。然而，悲剧的是我本地测算的结果显示，这个而所谓的优化版反而比原来的慢100倍。不过，还是把这个自定义收集器列出来。值得铭记的是，这个收集器是有序的，所以不能并行，那个这个combiner方法可以不要的，最好返回UnsupportedOperationException来警示此收集器的非并行性。

测试见 https://github.com/Ryan-Miao/l4Java/blob/master/src/test/java/com/test/java/stream/collect/PrimeNumbersCollectorTest.java

public class PrimeNumbersCollector implements    Collector
    
     >, Map
     
      >> {    @Override    public Supplier
      
     
    

参考

• Java8 in Action