Stream API

上篇内容我们学习了Stream的大部分终端操作，我们这篇着重了解下Stream中重要的终端操作：collect。

collect 方法

序号	支持的类	方法定义	方法说明
1	Stream<T>	<R> R collect(Supplier<R> supplier, BiConsumer<R, ? super T> accumulator, BiConsumer<R, R> combiner);	对此流的元素执行 mutable reduction操作。
2	Stream<T>	<R, A> R collect(Collector<? super T, A, R> collector);	使用 Collector对此流的元素执行 mutable reduction Collector。

以下代码见 StreamTerminalOperationTransformTest。

实现3参数转换接口

序号1的方法，传递了3个参数，参数1为创建新结果容器的函数；参数2为累加器函数，将参数1和流内元素执行累加操作；参数3为组合器函数，并行执行时会使用该函数。

同步执行时，该方法相当于执行：

R result = supplier.get();
for (T element : this stream) {
  accumulator.accept(result, element);
}
return result;

我们编写如下代码，看下实际效果

// 使用collect方法实现字符串连接
log.info("拼接字符串为：{}",
        Stream.of("I", "love", "you", "too")
                .collect(StringBuilder::new, (b1, b2) -> {
                    log.info("累加执行：{} + {}", b1, b2);
                    b1.append(b2);
                }, (b1, b2) -> {
                    log.info("组合执行：{} ++ {}", b1, b2);
                    b1.append(b2);
                })
                .toString());

以上代码将输出如下日志：

[main] INFO top.todev.note.web.flux.stream.StreamTerminalOperationTransformTest - 累加执行： + I

[main] INFO top.todev.note.web.flux.stream.StreamTerminalOperationTransformTest - 累加执行：I + love

[main] INFO top.todev.note.web.flux.stream.StreamTerminalOperationTransformTest - 累加执行：Ilove + you

[main] INFO top.todev.note.web.flux.stream.StreamTerminalOperationTransformTest - 累加执行：Iloveyou + too

[main] INFO top.todev.note.web.flux.stream.StreamTerminalOperationTransformTest - 拼接字符串为：Iloveyoutoo

并行执行时，该方法相当于执行：

R result1 = supplier.get();
R result2 = supplier.get();
R result3 = supplier.get();
R result4 = supplier.get();

// 累加执行，此处为并发（多线程）执行，每行代表一个线程
accumulator.accept(result1, element1);
accumulator.accept(result2, element2);
accumulator.accept(result3, element3);
accumulator.accept(result4, element4);
// ...
// accumulator.accept(resultN, elementN);

// 开始组合，此处为并发（多线程）执行，每行代表一个线程
combiner.accept(result1, result2);
combiner.accept(result3, result4);
combiner.accept(result1, result3);
// combiner.accept(result1, resultN);

return result1;

将上述的代码改为.parallel()方式调用，将输出如下日志：

[main] INFO top.todev.note.web.flux.stream.StreamTerminalOperationTransformTest - 累加执行： + you

[ForkJoinPool.commonPool-worker-3] INFO top.todev.note.web.flux.stream.StreamTerminalOperationTransformTest - 累加执行： + I

[ForkJoinPool.commonPool-worker-2] INFO top.todev.note.web.flux.stream.StreamTerminalOperationTransformTest - 累加执行： + too

[ForkJoinPool.commonPool-worker-2] INFO top.todev.note.web.flux.stream.StreamTerminalOperationTransformTest - 组合执行：you ++ too

[ForkJoinPool.commonPool-worker-1] INFO top.todev.note.web.flux.stream.StreamTerminalOperationTransformTest - 累加执行： + love

[ForkJoinPool.commonPool-worker-1] INFO top.todev.note.web.flux.stream.StreamTerminalOperationTransformTest - 组合执行：I ++ love

[ForkJoinPool.commonPool-worker-1] INFO top.todev.note.web.flux.stream.StreamTerminalOperationTransformTest - 组合执行：Ilove ++ youtoo

[main] INFO top.todev.note.web.flux.stream.StreamTerminalOperationTransformTest - 拼接字符串为：Iloveyoutoo

注意：上述日志中出现的ForkJoinPool.commonPool-worker-N为并发（多线程）执行时的线程名。

实现Collector接口

实现Collector需要实现如下4个接口：

// 一个创建并返回一个新的可变结果容器的函数。
Supplier<A> supplier();
// 将值折叠成可变结果容器的函数。
BiConsumer<A, T> accumulator();
// 一个接受两个部分结果并将其合并的函数。 
BinaryOperator<A> combiner();
// 执行从中间累积类型 A到最终结果类型 R的最终 R 。 
Function<A, R> finisher();
// 返回一个 Collector.Characteristics 类型的Set, 表示该收集容器的特征。
Set<Characteristics> characteristics();

collect方法执行时，他们的调用流程如下：

创建新的结果容器（supplier()）
将新的数据元素并入结果容器（accumulator()）
将两个结果容器组合成一个（combiner()）
在容器上执行可选的最终变换（finisher()）

简单来讲，生成容器A，通过accumulator针对A及流元素T执行累加，（如果并行存在的话）对多个A执行组合combiner，最终执行finisher后由A转换为R。对于使用者来说，A为中间变量，无关其实现细节。

我们实现一个计算整数流的平均数的Collector，代码如下：

// 使用collector实现求ping均值
log.info("[1, 2, 3, 4, 5, 6]的平均值：{}",
        Stream.of(1, 2, 3, 4, 5, 6)
                .parallel()
                .collect(new Collector<Integer, long[], Double>() {
                             @Override
                             public Supplier<long[]> supplier() {
                                 return () -> new long[2];
                             }

                             @Override
                             public BiConsumer<long[], Integer> accumulator() {
                                 return (a, t) -> {
                                     log.info("{}累加{}", a, t);
                                     a[0] += t;
                                     a[1]++;
                                 };
                             }

                             @Override
                             public BinaryOperator<long[]> combiner() {
                                 return (a, b) -> {
                                     log.info("{}组合{}", a, b);
                                     a[0] += b[0];
                                     a[1] += b[1];
                                     return a;
                                 };
                             }

                             @Override
                             public Function<long[], Double> finisher() {
                                 return (a) -> a[1] == 0 ? 0 : new Long(a[0]).doubleValue() / a[1];
                             }

                             @Override
                             public Set<Characteristics> characteristics() {
                                 Set<Characteristics> set = new HashSet<>();
                                 set.add(Characteristics.CONCURRENT);
                                 return set;
                             }
                         }
                )
);

常用Collector

通过上面的示例，我们实现了一个自定义的Collector，我们发现实现一个自定义的Collector还是比较麻烦的，需要实现5个接口。
Java 开发者们已经想到了这个问题，他们额外提供了一个 of 方法，可以通过lambda的方式创建 collector，类似 collect 中传递几个参数：提供者、累加器、组合器、完成器以及特征配置，此处我们就不细讲了。
Java 开发者们更为贴心的为我们创建了一些常用的 Collector ，让我们可以直接使用。这些常用的 Collector 实现放在 Collectors 类下，我们来了解下。

统计平均值 averagingXxx 的使用

Collectors 提供了 averagingDouble、averagingLong、averagingInt 3种统计平均值的 Collector 实现类，以下代码以 averagingInt 为例，由于使用方式相似，我们就不举例了。

// 使用collector实现求均值
log.info("[1, 2, 3, 4, 5, 6]的平均值：{}",
        Stream.of(1, 2, 3, 4, 5, 6)
        .collect(Collectors.averagingInt(n -> n))
);

统计元素个数 counting 的使用

该方法和 Stream 中的 count 方法一样。

// 使用collector获取元素数量
log.info("[1, 2, 3, 4, 5, 6]的个数：{}",
        Stream.of(1, 2, 3, 4, 5, 6)
                .collect(Collectors.counting())
);

统计总和 summingXxx 的使用

Collectors 提供了 summingDouble、summingLong、summingInt 3种统计求和值的 Collecto r实现类，同时还提供了 summarizingDouble 、 summarizingLong 、summarizingInt 3种统计对象的 Collector 实现类，以下代码以 summingInt 为例，由于使用方式相似，我们就不举例了。

// 使用collector获取总和
log.info("[1, 2, 3, 4, 5, 6]的总和：{}",
        Stream.of(1, 2, 3, 4, 5, 6)
                .collect(Collectors.summingInt(n -> n))
);

统计最小元素 minBy 的使用

// 使用collector获取最小元素
log.info("[1, 2, 3, 4, 5, 6]的最小值：{}",
        Stream.of(1, 2, 3, 4, 5, 6)
                .collect(Collectors.minBy(Integer::min))
                .get()
);

统计最大元素 maxBy 的使用

// 使用collector获取最da元素
log.info("[1, 2, 3, 4, 5, 6]的最大值：{}",
        Stream.of(1, 2, 3, 4, 5, 6)
                .collect(Collectors.maxBy(Integer::max))
                .get()
);

统计累加处理 reducing 的使用

reducing 和 Stream 中的 reduce 操作方法类似，我们就不详述了。

// 使用collector实现求均值
log.info("[1, 2, 3, 4, 5, 6]的求和：{}",
        Stream.of(1, 2, 3, 4, 5, 6)
                .collect(Collectors.reducing(0, Integer::sum))
);

转换映射 mapping 的使用

mapping 支持将第一个参数的结果再次执行转换，即向下游传递。

log.info("[1, 2, 3, 4, 5, 6]每个增加20后的平均值：{}",
        Stream.of(1, 2, 3, 4, 5, 6)
                .collect(Collectors.mapping(n -> n + 20, Collectors.averagingInt(n -> n)))
);

转换连接 joining 的使用

joining 提供了 3 种重载方法，支持传递分隔符、前缀、后缀等。

// 使用collector连接字符串
log.info("连接字符串为：{}",
        Stream.of("I", "love", "you", "too")
                .collect(Collectors.joining(" ", "Java, ", "!"))
);

转换为集合 toList 的使用

log.info("[1, 2, 3, 4, 5, 6, 5, 3, 6]转换为集合：{}",
        Stream.of(1, 2, 3, 4, 5, 6, 5, 3, 6)
                .collect(Collectors.toList())
);

转换为Map toMap 的使用

toMap 提供了 3 种重载方法，除了指定 Key 和 Value 的生成器外，区别在于对于 Key 重复时， Value的处理方式；以及初始Map的生成器。

log.info("[1, 2, 3, 4, 5, 6, 5, 3, 6]转换为Map：{}",
        Stream.of(1, 2, 3, 4, 5, 6, 5, 3, 6)
                .collect(Collectors.toMap(Object::toString, n -> n, Integer::sum))
);

转换为Set toSet 的使用

log.info("[1, 2, 3, 4, 5, 6, 5, 3, 6]的转换为Set：{}",
        Stream.of(1, 2, 3, 4, 5, 6, 5, 3, 6)
                .collect(Collectors.toSet())
);

转换为分组 groupingBy 的使用

分组函数将流中元素按某种定义分组，也提供了 2 种重载方法，支持递归向下游分组。

log.info("[1, 2, 3, 4, 5, 6, 5, 3, 6]的分组数据：{}",
        Stream.of(1, 2, 3, 4, 5, 6, 5, 3, 6)
                .collect(Collectors.groupingBy(n -> n))
);

转换为分区 partitioningBy 的使用

分区函数将流中元素按条件分为2组，也提供了 2 种重载方法，支持递归向下游分组。

log.info("[1, 2, 3, 4, 5, 6]的奇偶分区数据：{}",
        Stream.of(1, 2, 3, 4, 5, 6)
                .collect(Collectors.partitioningBy(n -> n %2 == 0))
);

其他方法

Collectors 中还提供了 groupingByConcurrent 、 toCollection 、 toConcurrentMap 等几种支持并发的 Collector 实现，用法基本和非并发的相同，我们就不详述了。

源码详见：https://github.com/crystalxmumu/spring-web-flux-study-note

以上是本期笔记的内容，我们下期见。

学习Java 8 Stream Api (4) - Stream 终端操作之 collect