java8新特性-引用流-max，min

例子：

  List<User> users = new ArrayList<>();
    users.add(new User("张三",30));
    users.add(new User("李四",34));
    users.add(new User("王五",20));

   
    Optional<User> max = users.stream().max(Comparator.comparing(User::getAge));
    System.out.println(max.get());

上面的例子求出最大年龄的User。

Stream的执行流程参考https://www.cnblogs.com/shigongp/p/17181380.html。下面只说max，min的处理逻辑。

源码分析

Comparator#comparing

    public static <T, U extends Comparable<? super U>> Comparator<T> comparing(
        Function<? super T, ? extends U> keyExtractor)
{
    Objects.requireNonNull(keyExtractor);
    return (Comparator<T> & Serializable)
        (c1, c2) -> keyExtractor.apply(c1).compareTo(keyExtractor.apply(c2));
}

按照传入的Function结果比较，上面的例子按照User::getAge比较。

ReferencePipeline#max(Comparator<? super P_OUT> comparator)

 @Override
public final Optional<P_OUT> max(Comparator<? super P_OUT> comparator) {
    return reduce(BinaryOperator.maxBy(comparator));
}

BinaryOperator#maxBy

 public static <T> BinaryOperator<T> maxBy(Comparator<? super T> comparator) {
    Objects.requireNonNull(comparator);
    return (a, b) -> comparator.compare(a, b) >= 0 ? a : b;
}

BinaryOperator#maxBy返回两个比较器中大的结果。

ReferencePipeline#reduce

  public final Optional<P_OUT> reduce(BinaryOperator<P_OUT> accumulator) {
    return evaluate(ReduceOps.makeRef(accumulator));
}

evaluate的执行参考https://www.cnblogs.com/shigongp/p/17181380.html。

ReduceOps#makeRef

public static <T> TerminalOp<T, Optional<T>>
makeRef(BinaryOperator<T> operator) {
    Objects.requireNonNull(operator);
    class ReducingSink
            implements AccumulatingSink<T, Optional<T>, ReducingSink> {
        private boolean empty;
        private T state;

        public void begin(long size) {
            empty = true;
            state = null;
        }

        @Override
        public void accept(T t) {
            if (empty) {
                empty = false;
                state = t;
            } else {
                state = operator.apply(state, t);
            }
        }

        @Override
        public Optional<T> get() {
            return empty ? Optional.empty() : Optional.of(state);
        }

        @Override
        public void combine(ReducingSink other) {
            if (!other.empty)
                accept(other.state);
        }
    }
    return new ReduceOp<T, Optional<T>, ReducingSink>(StreamShape.REFERENCE) {
        @Override
        public ReducingSink makeSink() {
            return new ReducingSink();
        }
    };
}

关注ReducingSink。

ReducingSink#begin

  public void begin(long size) {
            empty = true;
            state = null;
        }

设置empty为true，表示没有值，state为null，将值设为null。

ReducingSink#accept

  public void accept(T t) {
            if (empty) {
                empty = false;
                state = t;
            } else {
                state = operator.apply(state, t);
            }
        }

当执行数据处理调用Sink的accept时，调用ReducingSink#accept，按照处理数据的分割器，依次遍历元素，第一次进入此方法，empty为true，将empty设置为false，state设置为初始值，以后每次调用将值传入与state比较，operator是BinaryOperator.maxBy时表示取最大值，operator是BinaryOperator.minBy时取最小值。调用ReducingSink#get获取结果。