jdk1.8的stream学习之三
reduce函数 1：reduce(BinaryOperator<P_OUT> accumulator)
reduce函数，有3个重载的，我们分析下内部是怎么实现的。
在stream类中，第一个reduce方法接收一个BinaryOperator的函数接口。这个函数接口继承自BiFunction，
@Override
public final Optional<P_OUT> reduce(BinaryOperator<P_OUT> accumulator) {
return evaluate(ReduceOps.makeRef(accumulator));
}
继续跟踪代码进入到ReduceOps类中，在makeRef方法中，定义了一个内部类ReducingSink，然后返回了ReduceOps的一个实例，且重新实现了makeSink这个方法。这个方法就干了一件事，实例化了ReducingSink对象。
整个计算的过程是由ReducingSink类中，beagin，accept，get，combine四个方法实现的。上图中ReduceOps.makeRef执行后返回了一个新的ReduceOp实例对象。我们继续跟踪代码进入到evaluate方法。
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();
}
};
}
在evaluate方法内部，就做了一件事，判断是不是并行流。分别调用不同的方法。这里我们用的普通的方法，那么执行的是evaluateSequential这个方法。这个方法有2个参数，第一个参数是传入一个AbstractPipeline对象，这里传入的是自己，第二个是通过操作标识来返回一个具体的Spliterator对象。
final <R> R evaluate(TerminalOp<E_OUT, R> terminalOp) {
assert getOutputShape() == terminalOp.inputShape();
if (linkedOrConsumed)
throw new IllegalStateException(MSG_STREAM_LINKED);
linkedOrConsumed = true;
return isParallel()
? terminalOp.evaluateParallel(this, sourceSpliterator(terminalOp.getOpFlags()))
: terminalOp.evaluateSequential(this, sourceSpliterator(terminalOp.getOpFlags()));
}
继续跟踪代码到evaluateSequential，在这个方法里面通过调用makeSink方法来创建一个新的ReducingSink对象，如果不懂回到第二张图就可以看下返回值，返回值里实现了这个方法。继续往下跟踪
@Override
public <P_IN> R evaluateSequential(PipelineHelper<T> helper,
Spliterator<P_IN> spliterator) {
return helper.wrapAndCopyInto(makeSink(), spliterator).get();
}
这个方法里面调用了又调用了copyInto方法，(有点绕对不对，不要紧，其实这里主要就是3个类对象之间的相互调用，这3个类是ReduceOp，)
@Override
final <P_IN, S extends Sink<E_OUT>> S wrapAndCopyInto(S sink, Spliterator<P_IN> spliterator) {
copyInto(wrapSink(Objects.requireNonNull(sink)), spliterator);
return sink;
}
在copyInto方法里面就能看到我们想要的东西了。这里的wrappedSink对象，就是前面的ReducingSink对象的实例。这里先调用begin方法。回到图2看begin干了什么，然后又调用了spliterator.forEachRemaining，这里面就是调用了ReducingSink中的accept方法。然后遍历流中对象，依次调用，最后完成后调用end方法。
final <P_IN> void copyInto(Sink<P_IN> wrappedSink, Spliterator<P_IN> spliterator) {
Objects.requireNonNull(wrappedSink);
if (!StreamOpFlag.SHORT_CIRCUIT.isKnown(getStreamAndOpFlags())) {
wrappedSink.begin(spliterator.getExactSizeIfKnown());
spliterator.forEachRemaining(wrappedSink);
wrappedSink.end();
}
else {
copyIntoWithCancel(wrappedSink, spliterator);
}
}
然后forEachRemaining中调用的就是ReducingSink中(图2)的哪个accept，也就是我们传入的行为。从这里我们可以看到，每次会将结果返回给state，
@Override
public void accept(T t) {
if (empty) {
empty = false;
state = t;
} else {
state = operator.apply(state, t);
}
}
如下代码所示，代表着对流中的元素进行求和操作。
Optional<Integer> reduce = Stream.of(99, 5, 47, 34, 2).reduce((a, b) -> a + b);
System.out.println(JSONObject.toJSONString(reduce.get()));
最后执行结果返回的是：
187
reduce(final P_OUT identity, final BinaryOperator<P_OUT> accumulator)
接收2个参数，第一个参数相当于起始值，第二个函数相当于执行的操作
Integer reduce = Stream.of(99, 5, 47, 34, 2).reduce(100, (a, b) -> a + b);
System.out.println(JSONObject.toJSONString(reduce));
输出结果如下：
287
那么问题来了，多的这个参数是在哪里初始化的呢？
不带初始值的：在调用begin时候初始化未null。
private T state;
public void begin(long size) {
empty = true;
state = null;
}
那么带初始值的 ，也是在begin中，只不过这里进行根据传入的参数初始化了state这个值
@Override
public void begin(long size) {
state = seed;
}
reduce(R identity, BiFunction<R, ? super P_OUT, R> accumulator, BinaryOperator<R> combiner)
第3个函数有3个参数。第一个参数是初始值，第2，3个参数是运算。通过对源码进行分析，第三个参数的行为是在ReducingSink 里面的combine里面调用的，但是前面非并行的流式执行并未调用过这个方法。我们可以去看看是否有在并行的方法里面进行过调用。
public static <T, U> TerminalOp<T, U>
makeRef(U seed, BiFunction<U, ? super T, U> reducer, BinaryOperator<U> combiner) {
Objects.requireNonNull(reducer);
Objects.requireNonNull(combiner);
class ReducingSink extends Box<U> implements AccumulatingSink<T, U, ReducingSink> {
@Override
public void begin(long size) {
state = seed;
}
@Override
public void accept(T t) {
state = reducer.apply(state, t);
}
@Override
public void combine(ReducingSink other) {
state = combiner.apply(state, other.state);
}
}
return new ReduceOp<T, U, ReducingSink>(StreamShape.REFERENCE) {
@Override
public ReducingSink makeSink() {
return new ReducingSink();
}
};
}
在并行方法调用里面，创建了一个ReduceTask对象，这个对象是ForkJoinTask的子类，里面通过多线程去并行的执行。具体原理不在这里详述，有专门讲forkjoin的章节去将。
@Override
public <P_IN> R evaluateParallel(PipelineHelper<T> helper,
Spliterator<P_IN> spliterator) {
return new ReduceTask<>(this, helper, spliterator).invoke().get();
}
在redueTask中，一个很重要的方法onCompletion这里面调用了combine,这里讲结果进行合并。这里需要注意的是，这个并行流和普通的流计算方式是不一样的。
@Override
public void onCompletion(CountedCompleter<?> caller) {
if (!isLeaf()) {
S leftResult = leftChild.getLocalResult();
leftResult.combine(rightChild.getLocalResult());
setLocalResult(leftResult);
}
// GC spliterator, left and right child
super.onCompletion(caller);
}
如下所示例子：
我们给定4个数字，2,3,6,8,在非并行的情况下，结算过程如下：从0开始
第一次运算：c=0,d=2 ,返回的是2
第二次运算：c=2,d=3 ,返回的是5
第三次运算：c=5,d=6 ,返回的是11
第四次运算：c=11,d=8 ,返回的是19
ArrayList<Integer> integers = Lists.newArrayList(2,3,6,8);
Integer reduce = integers.stream().reduce(0,(c,d)->{
System.out.println(c);
System.out.println(d);
System.out.println("-----------------");
return (c+d);
} , (a, b) -> a * b);
System.out.println(JSONObject.toJSONString(reduce));
输出结果如下：
0
2
-----------------
2
3
-----------------
5
6
-----------------
11
8
-----------------
19
如果在并行流的情况下，计算过程则完全不一样了。
并行流的计算过程如下，4个线程分别通过第二个参数的加法运算进行运算：
也就是说，第二个函数值参与了第一次的运算，之后的合并结果的操作都依赖于第二个函数。这个一定要注意，如果仅仅是想把普通流转换为并行流，则两个函数的运算必须一致。
image.png
ArrayList<Integer> integers = Lists.newArrayList(2,3,6,8);
Integer reduce = integers.parallelStream().reduce(0,(c,d)-> (c+d) , (a, b) -> a * b);
System.out.println(JSONObject.toJSONString(reduce));
最后的输出结果如下：
288
hadoop大数据处理依赖的是mapreduce，在我们的流式计算中，也可以通过map,reduce来进行一样的操作。
基本思路学会以后，工作中对list对象的处理则可以用这些函数去操作。
