天天看点
返回

【Java8新特性】——Stream的reduce及Collect使用方式

文章目录

  • ​​前言​​
  • ​​一、Reduce​​
  • ​​1.1一个参数的Reduce​​
  • ​​BiFunction​​
  • ​​BinaryOperator​​
  • ​​1.2二个参数的Reduce​​
  • ​​1.3三个参数的Reduce​​
  • ​​非并行​​
  • ​​并行​​
  • ​​二、Collect​​
  • ​​BiConsumer​​
  • ​​三、Collector​​
  • ​​四、定制收集器​​
  • ​​总结​​

前言

本文主要讲解关于Stream中reduce的使用方式以及Collect使用方式，同时展示如何自定义收集器。

提示：如果大家对lambda表达式中的四大基础函数不清楚，推荐大家优先看下四大内置核心函数式接口​以及看下关于reduce相关api的使用，Java8 中reduce的基本使用

一、Reduce

Reduce中文含义为：减少、缩小；而Stream中的Reduce方法干的正是这样的活：根据一定的规则将Stream中的元素进行计算后返回一个唯一的值。

它有三个变种，输入参数分别是一个参数、二个参数以及三个参数；

1.1一个参数的Reduce 

Optional<T> reduce(BinaryOperator<T> accumulator)

先解释基本概念

BiFunction 

R apply(T t, U u);

函数式接口与Function不同点在于它接收两个输入返回一个输出；Function接收一个输入返回一个输出。两个输入，一个输出的类型可以不同。

BinaryOperator 

public interface BinaryOperator<T> extends BiFunction<T,T,T>

继承BiFunction，相比BinaryOperator直接限定其三个参数必须一样，表示两个相同类型的输入经过计算后产生一个同类型的输出。

BinaryOperator接口，可以看到reduce方法接受一个函数，这个函数有两个参数，第一个参数是上次函数执行的返回值（也称为中间结果），第二个参数是stream中的元素，函数将两个值按照方法处理，得到值赋给下次执行这个函数的参数。第一次执行的时候第一参数的值是stream的第一元素，第二个元素是stream的第二元素，因为stream元素集合可能为空，所以这个方法的返回值为Optional。举个例子感受一下

Optional accResult = Stream.of(1, 2, 3)
        .reduce((acc, item) -> {
            System.out.println("acc : "  + acc);
            acc += item;
            System.out.println("item: " + item);
            System.out.println("acc+ : "  + acc);
            System.out.println("--------");
            return acc;
        });
System.out.println("accResult: " + accResult.get());
System.out.println("--------");
// 结果打印
--------
acc : 1
item: 2
acc+ : 3
--------
acc : 3
item: 3
acc+ : 6
--------
accResult: 6
--------

1.2二个参数的Reduce 

T reduce(T identity, BinaryOperator<T> accumulator);

与第一个变形不同的，会接受一个返回值类型相同identity参数，用于指定Stream的循环初始值，如果Stream为空，则默认返回identity，则不会再返回null值。

int accResult = Stream.of(1, 2, 3)
            .reduce(0, (acc, item) -> {
                System.out.println("acc : "  + acc);
                acc += item;
                System.out.println("item: " + item);
                System.out.println("acc+ : "  + acc);
                System.out.println("--------");
                return acc;
            });
System.out.println("accResult: " + accResult);
System.out.println("--------");
// 结果打印
acc : 0
item: 1
acc+ : 1
--------
acc : 1
item: 2
acc+ : 3
--------
acc : 3
item: 3
acc+ : 6
--------
accResult: 6
--------

1.3三个参数的Reduce 

<U> U reduce(U identity,BiFunction<U, ? super T, U> accumulator,BinaryOperator<U> combiner);

解释过前两个值的概念，第三个值combiner作用，主要用于并发进行的 为了避免竞争 每个reduce线程都会有独立的result combiner的作用在于合并每个线程的result得到最终结果。如果Stream是非并行时，第三个参数实际上是不生效的。

但是如果Stream是并行，第三个参数有了意义，将不同线程计算的结果调用combiner做汇总后返回。

.out.println("----------stream---------");
        System.out.println(Stream.of(1, 2, 3).reduce(4, (s1, s2) -> s1 + s2
                , (s1, s2) -> s1 + s2));
        
        System.out.println("----------parallel stream---------");
        System.out.println(Stream.of(1, 2, 3).parallel().reduce(4, (s1, s2) -> s1 + s2
                , (s1, s2) -> s1 + s2));

----------stream---------
10
----------parallel stream---------
18

此时看下非并行和并行的打印结果，运算结果并不一样，原因在哪？

非并行

计算过程：第一步计算4 + 1 = 5，第二步是5 + 2 = 7，第三步是7 + 3 = 10。按非并行的方式来看它是分了三步的，每一步都要依赖前一步的运算结果。

并行

计算过程：并行计算时，线程之间没有影响，因此每个线程在调用第二个参数BiFunction进行计算时，直接都是使用result值当其第一个参数（由于Stream计算的惰性处理，在调用最终方法前，都不会进行实际的运算，因此每个线程取到的result值都是原始的4），因此计算过程现在是这样的：线程1：1 + 4 = 5；线程2：2 + 4 = 6；线程3：3 + 4 = 7；Combiner函数： 5 + 6 + 7 = 18

二、Collect

collect就是收集器，是Stream一种通用的、从流生成复杂值的结构。只要将它传给collect方法，也就是所谓的转换方法，其就会生成想要的数据结构。

<R> R collect(Supplier<R> supplier,
        BiConsumer<R, ? super T> accumulator,
        BiConsumer<R, R> combiner);
  • supplier：动态的提供初始化的值；创建一个可变的结果容器（JAVADOC）；对于并行计算，这个方法可能被调用多次，每次返回一个新的对象；
  • accumulator：类型为BiConsumer，注意这个接口是没有返回值的；它必须将一个元素放入结果容器中（JAVADOC）。
  • combiner：类型也是BiConsumer，因此也没有返回值。它与三参数的Reduce类型，只是在并行计算时汇总不同线程计算的结果。它的输入是两个结果容器，必须将第二个结果容器中的值全部放入第一个结果容器中（JAVADOC）。

BiConsumer 

public interface BiConsumer<T, U> {
    void accept(T t, U u);
}

可见它就是一个两个输入参数的Consumer的变种。计算没有返回值。即消费型。

与reduce一样，同样分为并行和非并行，下面讲解下并行

/**
 * 模拟Filter查找其中含有字母a的所有元素，打印结果将是aa ab ad
 */
Stream<String> s1 = Stream.of("aa", "ab", "c", "ad");
Predicate<String> predicate = t -> t.contains("a");
System.out.println(s1.parallel().collect(() -> new ArrayList<String>(),
    (array, s) -> {if (predicate.test(s)) array.add(s); },
    (array1, array2) -> array1.addAll(array2)));

每个线程都创建了一个结果容器ArrayList，假设每个线程处理一个元素，那么处理的结果将会是[aa],[ab],[],[ad]四个结果容器（ArrayList）；最终再调用第三个BiConsumer参数将结果全部Put到第一个List中，因此返回结果就是打印的结果了。

三、Collector

Collector是Stream的可变减少操作接口，可变减少操作包括：将元素累积到集合中，使用StringBuilder连接字符串;计算元素相关的统计信息，例如sum，min，max或average等。Collectors这个工具库，在该库中封装了相应的转换方法。当然，Collectors工具库仅仅封装了常用的一些情景，如果有特殊需求，那就要自定义了。

Collector<T, A, R>接受三个泛型参数，对可变减少操作的数据类型作相应限制：

T：输入元素类型
A：缩减操作的可变累积类型（通常隐藏为实现细节）
R：可变减少操作的结果类型

Collector接口声明了4个函数，这四个函数一起协调执行以将元素目累积到可变结果容器中，并且可以选择地对结果进行最终的变换.

Supplier<A> supplier(): 创建新的结果结
BiConsumer<A, T> accumulator(): 将元素添加到结果容器
BinaryOperator<A> combiner(): 将两个结果容器合并为一个结果容器
Function<A, R> finisher(): 对结果容器作相应的变换
Collector接口声明了4个函数，这四个函数一起协调执行以将元素目累积到可变结果容器中，并且可以选择地对结果进行最终的变换.

在Collector接口的characteristics方法内，可以对Collector声明相关约束
Set<Characteristics> characteristics():

四、定制收集器

定制收集器需要实现Collector接口

实现Collector接口需要给定三个泛型

第一个泛型：收集元素的类型

第二个泛型：累加器的类型

第三个泛型：最终结果的类型

static class CollectorImpl<T, A, R> implements Collector<T, A, R> {
    private final Supplier<A> supplier;
    private final BiConsumer<A, T> accumulator;
    private final BinaryOperator<A> combiner;
    private final Function<A, R> finisher;
    private final Set<Characteristics> characteristics;

    CollectorImpl(Supplier<A> supplier,
                  BiConsumer<A, T> accumulator,
                  BinaryOperator<A> combiner,
                  Function<A,R> finisher,
                  Set<Characteristics> characteristics) {
        this.supplier = supplier;
        this.accumulator = accumulator;
        this.combiner = combiner;
        this.finisher = finisher;
        this.characteristics = characteristics;
    }

    CollectorImpl(Supplier<A> supplier,
                  BiConsumer<A, T> accumulator,
                  BinaryOperator<A> combiner,
                  Set<Characteristics> characteristics) {
        this(supplier, accumulator, combiner, castingIdentity(), characteristics);
    }

    @Override
    public BiConsumer<A, T> accumulator() {
        return accumulator;
    }

    @Override
    public Supplier<A> supplier() {
        return supplier;
    }

    @Override
    public BinaryOperator<A> combiner() {
        return combiner;
    }

    @Override
    public Function<A, R> finisher() {
        return finisher;
    }

    @Override
    public Set<Characteristics> characteristics() {
        return characteristics;
    }
}

对应着Collector中四个流程来完成

Supplier<A> supplier(): 创建新的结果结
BiConsumer<A, T> accumulator(): 将元素添加到结果容器
BinaryOperator<A> combiner(): 将两个结果容器合并为一个结果容器
Function<A, R> finisher():

第一步生成新的结果集Supplier supplier(): 创建新的结果结

public Supplier<StringJoiner> supplier() {
    return () -> new StringJoiner(delim, prefix, suffix);
}
【Java8新特性】——Stream的reduce及Collect使用方式

第二步结合之前操作的结果和当前值，生成并返回新的值，即BiConsumer<A, T> accumulator()

public BiConsumer<StringJoiner, String> accumulator() {
        return StringJoiner::add;
    }
【Java8新特性】——Stream的reduce及Collect使用方式

第三步将两个容器合并，由于Collector支持并发操作，如果不将多的容器合并，必然会导致数据的混乱。如果仅仅在串行执行，此步骤可以省略。即BinaryOperator combiner()

public BinaryOperator<StringJoiner> combiner() {
        return StringJoiner::merge;
    }
【Java8新特性】——Stream的reduce及Collect使用方式

第四步处理返回值，即Function<A, R> finisher(): 对结果容器作相应的变换。

public Function<StringJoiner, String> finisher() {
        return StringJoiner::toString;
    }

Collector自定义起来，也不是特别的麻烦，不过要明确以下几点：

  1. 参数类型：这里最重要的是指定累加器的类型，一般都是自定义的过度类

    待收集元素的类型：T；

    累加器的类型：A；

    最终结果的类型：R。

  2. 累加器的逻辑
  3. 最终结果的转换
  4. Collector特征的选择

总结

java8 reduce ico ide 泛型
上一篇: CDH 6.3.1 设置邮件告警功能【2020年06月】
下一篇: Hue: Thrift Server role must be configured in HBase service to use the Hue HBase Browser application

继续阅读