Java8-理解Collector

内容预览:
  • @Overridepublic Supplier<List<T>> supplier() { return A...~
  • 当遍历到流中第n个元素时,这个函数就会执行~
  • 而ArrayList这种队列一样的数据结构则生成有序的stream,不能并行~

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

Collector定义

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

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

Collector接口定义了5个函数

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如下:

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

理解Collector几个函数

建立新的结果容器 supplier方法

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

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

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

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

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

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

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

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

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

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

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

顺序归约

合并两个结果容器 combiner

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

@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方法就好。

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

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

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

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

以上就是:Java8-理解Collector 的全部内容。

本站部分内容来源于互联网和用户投稿,如有侵权请联系我们删除,谢谢。
Email:[email protected]


0 条回复 A 作者 M 管理员
    所有的伟大,都源于一个勇敢的开始!
欢迎您,新朋友,感谢参与互动!欢迎您 {{author}},您在本站有{{commentsCount}}条评论