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

Collector定义

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

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

Collector接口定义了5个函数

1
2
3
4
5
6
7
public interface Collector<T, A, R> {
    Supplier<A> supplier();
    BiConsumer<A, T> accumulator();
    BinaryOperator<A> combiner();
    Function<A, R> finisher();
    Set<Characteristics> characteristics();
}
  1. T是流中要收集的对象的泛型
  2. A是累加器的类型，累加器是在收集过程中用于累积部分结果的对象。
  3. R是收集操作得到的对象(通常但不一定是集合)的类型。

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

1
public class ToListCollector<T> implements Collector<T, List<T>, List<T>>

理解Collector几个函数

建立新的结果容器 supplier方法

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

1
2
3
4
@Override
public Supplier<List<T>> supplier() {
    return ArrayList::new;
}

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

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

1
2
3
4
@Override
public BiConsumer<List<T>, T> accumulator() {
    return List::add;
}

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

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

1
2
3
4
@Override
public Function<List<T>, List<T>> finisher() {
    return (i) -> i;
}

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

1
2
3
static <T> Function<T, T> identity() {
    return t -> t;
}

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

顺序归约

合并两个结果容器 combiner

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

1
2
3
4
5
6
7
8
@Override
public BinaryOperator<List<T>> 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方法就好。

1
2
3
List<Dish> rs = dishes
            .stream()
            .collect(new ToListCollector<>());

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

1
2
3
4
5
6
7
List<Dish> 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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
public class PrimeNumbersCollector implements
    Collector<Integer, Map<Boolean, List<Integer>>, Map<Boolean, List<Integer>>> {

    @Override
    public Supplier<Map<Boolean, List<Integer>>> supplier() {
        return () -> {
            Map<Boolean, List<Integer>> map = new HashMap<>();
            map.put(true, new ArrayList<>());
            map.put(false, new ArrayList<>());
            return map;
        };
    }

    @Override
    public BiConsumer<Map<Boolean, List<Integer>>, Integer> accumulator() {
        return (Map<Boolean, List<Integer>> acc, Integer candidate) -> {
            acc.get(isPrime(acc.get(true), candidate)).add(candidate);
        };
    }

    /**
     * 从质数列表里取出来，看看是不是candidate的约数.
     *
     * @param primes 质数列表
     * @param candidate 判断值
     * @return true -> 质数； false->非质数。
     */
    private static Boolean isPrime(
        List<Integer> primes,
        Integer candidate) {
        int candidateRoot = (int) Math.sqrt((double) candidate);
        return primes.stream().filter(p -> p<=candidateRoot).noneMatch(i -> candidate % i == 0);
    }

    @Override
    public BinaryOperator<Map<Boolean, List<Integer>>> combiner() {
        return (Map<Boolean, List<Integer>> map1, Map<Boolean, List<Integer>> map2) -> {
            map1.get(true).addAll(map2.get(true));
            map1.get(false).addAll(map2.get(false));
            return map1;
        };
    }

    @Override
    public Function<Map<Boolean, List<Integer>>, Map<Boolean, List<Integer>>> finisher() {
        return Function.identity();
    }

    @Override
    public Set<Characteristics> characteristics() {
        return Collections.unmodifiableSet(EnumSet.of(Characteristics.IDENTITY_FINISH));
    }
}

参考

  • Java8 in Action