JAVA相关

Stream 学习笔记（中）

2018-08-13  本文已影响171人  freeseawind
开发环境
前置知识点
关于 Reduction

Stream包含许多最终操作，比如: average、sum、min、max、count，它们通过通过组合流的内容返回一个值。这些操作我们暂时称之为 “统计操作“（Reduction）。

下面我们来分别介绍两种不同的 “统计操作“ 方式：

关于Reduction的命名，有更好的提议可以留言告知。

Stream.reduce 方法

如下所示，该管道计算集合中男性成员年龄的总和，它使用 Stream.sum 方法来进行 “统计操作“

Integer totalAge = roster.stream().mapToInt(Person::getAge).sum();

使用Stream.reduce操作来计算相同的值

roster.stream().map(Person::getAge).reduce(0, (a, b) -> a + b).intValue();
Stream.reduce方法包含两个参数：

如果reduce操作涉及向集合添加元素，那么每次accumulator函数处理元素时，它都会创建一个包含元素的新集合，这可能会影响性能。

Stream.collect 方法

与reduce方法不同，reduce方法在处理元素时始终创建新值，而collect方法修改或改变现有值。
考虑如何计算平均值，需要两类数据：总记录数和记录的总和。collect方法只创建一次新值，用来跟踪总记录数和记录的总和，例如以下类Averager：

static class Averager implements IntConsumer
{
    private int total = 0;
    private int count = 0;

     public double average()
     {
          return count > 0 ? ((double) total) / count : 0;
      }

      public void accept(int i)
      {
          total += i;
          count++;
      }

      public void combine(Averager other)
      {
          total += other.total;
          count += other.count;
      }
 }

如下所示，该管道使用collect方法计算集合中男性成员年龄的总和

// collect
Averager averageCollect = roster.stream().filter(p -> p.getGender() == Person.Sex.MALE).map(Person::getAge).
collect(Averager::new, Averager::accept, Averager::combine);

 System.out.println("Average age of male members: " + averageCollect.average());
Stream.collect 方法包含三个参数：
Stream.collect 方法的其它用例
过滤结果集
List<Integer> list = roster.stream()
.filter(p -> p.getGender() == Person.Sex.MALE). map(Person::getAge).collect(Collectors.toList());
按关键字分组
Map<Person.Sex, List<Person>> byGender =
    roster
        .stream()
        .collect(
            Collectors.groupingBy(Person::getGender));
分组汇总
Map<Person.Sex, Double> averageAgeByGender = roster
    .stream()
    .collect(
        Collectors.groupingBy(
            Person::getGender,                      
            Collectors.averagingInt(Person::getAge)));
源码学习

查看下面代码的完整调用示例

Map<Person.Sex, Double> averageAgeByGender0 = roster.stream()
                .collect(Collectors.groupingBy(Person::getGender, Collectors.averagingInt(Person::getAge)));
Collector 部分源码
public interface Collector<T, A, R> {

    Supplier<A> supplier();

    BiConsumer<A, T> accumulator();

    BinaryOperator<A> combiner();

    Function<A, R> finisher();
}
通过Collector接口声明，我们可以看到几个关键的方法申明：
泛型说明
Collectors.averagingInt 源码
public static <T> Collector<T, ?, Double>
    averagingInt(ToIntFunction<? super T> mapper) {
        return new CollectorImpl<>(
                () -> new long[2],
                (a, t) -> { a[0] += mapper.applyAsInt(t); a[1]++; },
                (a, b) -> { a[0] += b[0]; a[1] += b[1]; return a; },
                a -> (a[1] == 0) ? 0.0d : (double) a[0] / a[1], CH_NOID);
}
通过Collector接口可知，Collectors.averagingInt方法是如何实现求平均值的
  1. 创建long数组保存中间计算结果
  2. 调用accumulator方法进行累计，并把中间结果存储到long数组中
  3. 如果需要合并多个中间结果，则把两个元素的结果进行汇总保存到第一个流元素中
  4. 计算并得到平均值
Collectors.groupingBy源码
public static <T, K, D, A, M extends Map<K, D>>
    Collector<T, ?, M> groupingBy(Function<? super T, ? extends K> classifier,
                                  Supplier<M> mapFactory,
                                  Collector<? super T, A, D> downstream) {
        Supplier<A> downstreamSupplier = downstream.supplier();
        BiConsumer<A, ? super T> downstreamAccumulator = downstream.accumulator();
        BiConsumer<Map<K, A>, T> accumulator = (m, t) -> {
            K key = Objects.requireNonNull(classifier.apply(t), "element cannot be mapped to a null key");
            A container = m.computeIfAbsent(key, k -> downstreamSupplier.get());
            downstreamAccumulator.accept(container, t);
        };
        BinaryOperator<Map<K, A>> merger = Collectors.<K, A, Map<K, A>>mapMerger(downstream.combiner());
        @SuppressWarnings("unchecked")
        Supplier<Map<K, A>> mangledFactory = (Supplier<Map<K, A>>) mapFactory;

        if (downstream.characteristics().contains(Collector.Characteristics.IDENTITY_FINISH)) {
            return new CollectorImpl<>(mangledFactory, accumulator, merger, CH_ID);
        }
        else {
            @SuppressWarnings("unchecked")
            Function<A, A> downstreamFinisher = (Function<A, A>) downstream.finisher();
            Function<Map<K, A>, M> finisher = intermediate -> {
                intermediate.replaceAll((k, v) -> downstreamFinisher.apply(v));
                @SuppressWarnings("unchecked")
                M castResult = (M) intermediate;
                return castResult;
            };
            return new CollectorImpl<>(mangledFactory, accumulator, merger, finisher, CH_NOID);
        }
}
源码剖析

了解相应泛型对应本例的类型

探讨部分

探讨点一

# 入参部分
Supplier<M> mapFactory

# 后续又进行了强制类型转换
Supplier<Map<K, A>> mangledFactory = (Supplier<Map<K, A>>) mapFactory;

从源码可以看出 mapFactory工厂方法应返回Map<Person.Sex, Long[]> 对象，所以上述方法参数是否为如下会更加明了

public static <T, K, D, A, M extends Map<K, D>>
    Collector<T, ?, M> groupingBy(Function<? super T, ? extends K> classifier,
                                  Supplier<K, A> mapFactory,
                                  Collector<? super T, A, D> downstream)

探讨点二

Function<A, A> downstreamFinisher = (Function<A, A>) downstream.finisher();
Function<Map<K, A>, M> finisher = intermediate -> {
    intermediate.replaceAll((k, v) -> downstreamFinisher.apply(v));
    @SuppressWarnings("unchecked")
    M castResult = (M) intermediate;
    return castResult;
};

源码中终止方法又进行了两次强制类型转换

  1. 为了调用intermediate.replaceAll,对downstream方法进行了一次强制类型转换
  2. 为了返回类型约束的结果对M进行了一次强制类型转换

重构如下

# 修改方法的声明
public static <T, K, D, A> Collector<T, ?, Map<K, D>> groupingBy(....)

# 修改结果返回函数
@SuppressWarnings("unchecked")
Function<Map<K, A>, Map<K, D>> finisher = intermediate ->
{
    Map<K, D> castResult = new HashMap<>();

    intermediate.entrySet().stream()
        .forEach(t -> castResult.put(t.getKey(), downstream.finisher().apply(t.getValue())));
         return castResult;
};

以上仅为个人观点，水平有限，欢迎大家指正。

关于泛型，它的初衷是让具有强制类型转换的代码具有更好的安全性和可读性。

Github工程地址

上一篇下一篇

猜你喜欢

热点阅读