• 如果您觉得本站非常有看点,那么赶紧使用Ctrl+D 收藏吧

java8学习总结——Stream的理解

stream 来源:漠风- 1次浏览
    Stream初理解

    JAVA8中一个重要概念就是流——stream。先来看看流的使用:

public class StreamTest01 {

    public static void main(String[] args) {
       Stream<Integer> stream = Stream.of(1,2,3,4);

       stream.map(i -> i+1).forEach(System.out::println);
    }
}

     上面的代码是创建了一个包含(1,2,3,4)四个元素的流,并将流中的元素分别加一,然后打印输出。也许这样不太直观。可以再看看下面的写法:

public class StreamTest01 {

    public static void main(String[] args) {
       List<Integer> list = Arrays.asList(1,2,3,4);
       Stream<Integer> stream = list.stream();
       stream.map(i -> i+1).forEach(System.out::println);
    }
}

上面的这段代码也是将四个元素加一然后输出到控制台。不同的是这里是通过一个集合创建流对象,然后进行操作的。同样的需求,流对象带来的便利性是显而易见,直观、易读且易编写。开发人员不用关心元素是如何取出的,也不用关心操作之间应该如何连接(以前的写法是,我们必须保证加一操作和输出操作之间的执行顺序);开发人员只需要关心每个操作的具体执行逻辑。

       文档将对其的解释是:流是一些聚合操作,可以并发和顺序执行。从这个定义中可以看出流的两个特征:

       1、流是一系列操作的集合。

       2、流可以并发执行,也可以顺序执行。

既然流是一些操作的集合,那么流就可以将一个或多个操作聚集起来。这里就有两个问题:1、流是如何将这些操作聚集起来的;2、流是如何创建的的,只有知道流的创建过程,才知道流是如何连接操作的。

      上面用到了Stream.of()方法返回了一个流对象。意味着这个方法包含了流的创建过程,先看看他的源代码:

//Stream接口
public static<T> Stream<T> of(T... values) {
        return Arrays.stream(values);
}

//Arrays类
public static <T> Stream<T> stream(T[] array) {
        return stream(array, 0, array.length);
}

//Arrays类
public static <T> Stream<T> stream(T[] array, int startInclusive, int endExclusive) {
        return StreamSupport.stream(spliterator(array, startInclusive, endExclusive), false);
}

//StreamSupport类
public static <T> Stream<T> stream(Spliterator<T> spliterator, boolean parallel) {
        Objects.requireNonNull(spliterator);
        return new ReferencePipeline.Head<>(spliterator,
                                            StreamOpFlag.fromCharacteristics(spliterator),
                                            parallel);
}

上面是Stream.of()方法的源代码调用过程,可以看到最后调用了StreamSupport类的stream()方法,在stream方法中创建了一个ReferencePipeline.Head类(这是
ReferencePipeline的一个内部类)的对象。接下来再看看
Head这个内部类的源代码: 

//Head类
static class Head<E_IN, E_OUT> extends ReferencePipeline<E_IN, E_OUT> {      
        Head(Supplier<? extends Spliterator<?>> source,
             int sourceFlags, boolean parallel) {
            super(source, sourceFlags, parallel);
        } 
……
}

// ReferencePipeline类
abstract class ReferencePipeline<P_IN, P_OUT> extends AbstractPipeline<P_IN, P_OUT, Stream<P_OUT>>
        implements Stream<P_OUT>  {
    ReferencePipeline(Supplier<? extends Spliterator<?>> source,int sourceFlags, boolean parallel) {
        super(source, sourceFlags, parallel);
}
……
}

// AbstractPipeline类
/**
* Constructor for the head of a stream pipeline.
*
* @param source {@code Spliterator} describing the stream source
* @param sourceFlags the source flags for the stream source, described in
* {@link StreamOpFlag}
* @param parallel {@code true} if the pipeline is parallel
*/
abstract class AbstractPipeline<E_IN, E_OUT, S extends BaseStream<E_OUT, S>>
        extends PipelineHelper<E_OUT> implements BaseStream<E_OUT, S> {
        AbstractPipeline(Supplier<? extends Spliterator<?>> source,
                     int sourceFlags, boolean parallel) {
              this.previousStage = null;
              this.sourceSupplier = source;
              this.sourceStage = this;
              this.sourceOrOpFlags = sourceFlags & StreamOpFlag.STREAM_MASK;
              // The following is an optimization of:
              // StreamOpFlag.combineOpFlags(sourceOrOpFlags, StreamOpFlag.INITIAL_OPS_VALUE);
              this.combinedFlags = (~(sourceOrOpFlags << 1)) & StreamOpFlag.INITIAL_OPS_VALUE;
              this.depth = 0;
              this.parallel = parallel;
        }
        ……
}

这里只截取了创建Head实例的部分代码,可以看到在Head的构造方法中什么也没做,直接调用了父类的构造方法。最后Head的构造方法调用的是AbstractPipeline类的构造方法。在
AbstractPipeline的构造方法执行完成后,流对象就创建完成了。

      但是另外一个问题又出现了。这么云山雾绕的转了一大圈的创建了一个对象,不知道是做什么用的。为什么要这么做,与流的创建又有什么关系?其实,在AbstractPipeline的构造方法我将注释也截取下来了,注释有这么一句话:

Constructor for the head of a stream pipeline(构造一个流管道的头部).这句话透露了两个信息:1、我们最后构造的是一个流管道,也就是说流对象是以管道的形式表现的;2、这个构造方法仅仅是构造管道的头部,也就是管道的开端,那么流管道的其他部分呢?

      因此,一定有构造流管道其他部分的方法的。这个方法就在AbstractPipeline类的另外一个构造方法里面,看一下AbstractPipeline类的另外一个构造方法:

 /**
 * Constructor for appending an intermediate operation stage onto an
 * existing pipeline.
 *
 * @param previousStage the upstream pipeline stage
 * @param opFlags the operation flags for the new stage, described in
 * {@link StreamOpFlag}
 */
AbstractPipeline(AbstractPipeline<?, E_IN, ?> previousStage, int opFlags) {
	if (previousStage.linkedOrConsumed)
		throw new IllegalStateException(MSG_STREAM_LINKED);
	previousStage.linkedOrConsumed = true;
	previousStage.nextStage = this;

	this.previousStage = previousStage;
	this.sourceOrOpFlags = opFlags & StreamOpFlag.OP_MASK;
	this.combinedFlags = StreamOpFlag.combineOpFlags(opFlags, previousStage.combinedFlags);
	this.sourceStage = previousStage.sourceStage;
	if (opIsStateful())
		sourceStage.sourceAnyStateful = true;
	this.depth = previousStage.depth + 1;
}

留意注释部分(Constructor for appending an intermediate operation stage onto an existing pipeline.)。根据注释的解释,这个构造方法构造的对象有两个特征:

1、是一个中间操作阶段对象(an  intermediate operation stage);

2、这个中间操作阶段对象会被追加到一个已经存在的管道上。

AbstractPipeline中有几个属性需要留意一下:

/**
 * Backlink to the head of the pipeline chain (self if this is the source
 * stage).
 */
@SuppressWarnings("rawtypes")
private final AbstractPipeline sourceStage;

/**
 * The "upstream" pipeline, or null if this is the source stage.
 */
@SuppressWarnings("rawtypes")
private final AbstractPipeline previousStage;

/**
 * The next stage in the pipeline, or null if this is the last stage.
 * Effectively final at the point of linking to the next pipeline.
 */
@SuppressWarnings("rawtypes")
private AbstractPipeline nextStage;

这三个属性都是
AbstractPipeline对象,分别表述pipeline的开端(sourceStage)、上一个操作(previousStage)、下一个操作(nextStage)。再结合
AbstractPipeline的两个构造方法看一下:

//构造方法一:构造管道的开端
this.previousStage = null;   //因为是开端,没有前一个状态
this.sourceStage = this;     //因为是开端,所以pipeline的源阶段就是当前对象

//构造方法二:构造流的中间操作
previousStage.nextStage = this;      //前一个操作的nextStage就是当前对象
this.previousStage = previousStage;  //当前操作的前一个操作就是previousStage
this.sourceStage = previousStage.sourceStage;  //无论管道有多长,其源都只会有一个(就像水管无论有多长,其开端只有一个),所以等于 previousStage.sourceStage

综上所述,我们可以得到流的结构是一种双向链表的管道结构,如下图:


流将每个操作封装成一个AbstractPipeline对象,然后通过previousStage、nextStage属性将当前操作的前一个操作和后一个操作连接起来,以此来保证流的将操作联合的特性。

        Stream的源
        Stream是一些列操作的集合,其需要一个操作的数据源,这个数据源就是Stream的源。在AbstractPipeline提供了两个构造流开端的方法,如下:

AbstractPipeline(Supplier<? extends Spliterator<?>> source,int sourceFlags, boolean parallel) { this.previousStage = null; this.sourceSupplier = source; this.sourceStage = this; …… } AbstractPipeline(Spliterator<?> source,int sourceFlags, boolean parallel) { this.previousStage = null; this.sourceSpliterator = source; this.sourceStage = this; …… }

上面两个构造方法中有一个变量(source),这个就是外部传入的数据源。在两个构造方法中,这个源分别被赋给了sourceSupplier 、sourceSpliterator两个属性。其实两个属性是一个意思,sourceSupplier是supplier(不接受参数,返回一个值)的一个实例,其返回的也是一个Spliterator对象。只是提供两种不同的源提供方式,但是其提供的源都是一样的。所以官方文档对这两个属性的介绍是:两个属性被使用了其中一个,另外一个必须为null(也就是说这个两个属性是互斥的)。在流使用完成后,这两个属性都必须被设置为null(流使用完了,就不再需要源了)。

  从上面的描述中可以看到source对象是一个Spliterator的实例。来看看Spliterator的源代码:

public interface Spliterator<T> { boolean tryAdvance(Consumer<? super T> action); //消耗式获取元素,元素一旦被取出,分割器将不会拥有这个元素 default void forEachRemaining(Consumer<? super T> action) { //遍历元素 do { } while (tryAdvance(action)); } Spliterator<T> trySplit(); //分割数据源 default long getExactSizeIfKnown() { return (characteristics() & SIZED) == 0 ? -1L : estimateSize(); } default boolean hasCharacteristics(int characteristics) { return (characteristics() & characteristics) == characteristics; } default Comparator<? super T> getComparator() { throw new IllegalStateException(); } }

官方文档将Spliterator解释为:是一个能够对源数据进行分区和遍历操作的对象。从其实现类中,可以更好的理解这个接口。Spliterator的实现类分为三类:       1、 ArraySpliterator       ( IntArraySpliterator,LongArraySpliterator,DoubleArraySpliterator);       2、 AbstractSpliterator  IntAbstractSpliterator,LongAbstractSpliterator,DoubleAbstractSpliterator       3、 IteratorSpliterator    IntIteratorSpliterator,LongIteratorSpliterator,DoubleIteratorSpliterator       关于这个三个实现类:       ArraySpliterator主要面向数组类型的数据源;       IteratorSpliterator面向的是集合类数据源;       AbstractSpliterator是给使用者自定义的,使用者可以根据自己需要扩展相应的Spliterator,这个类没有给出数据源类型,需要使用者自定义。       这个三个类的实现方式很类似,都有一些共同的特点:1、持有一个数据源;2、设置了数据源的大小、切分单位、切分的初始值;3、定义了分割器的特征(这里不做讨论)。从源代码中看看这些特征:

public static abstract class AbstractSpliterator<T> implements Spliterator<T> { static final int BATCH_UNIT = 1 << 10; // 切分数据源的单位,每次切分的增量。即batch+BATCH_UNIT就是本次切分的长度;这里默认是1024 static final int MAX_BATCH = 1 << 25; //允许切分的最大长度,即batch+BATCH_UNIT是当前切分的长度,这个长度不能大于MAX_BATCH。如果大于,则等于MAX_BATCH; private final int characteristics; //分割器的特征 private long est; // 数据源的长度(集合的长度) private int batch; // 切分的初始值,创建时一般是0 …… } static class IteratorSpliterator<T> implements Spliterator<T> { static final int BATCH_UNIT = 1 << 10; // 切分数据源的单位,每次切分的增量。即batch+BATCH_UNIT就是本次切分的长度;这里默认是1024 static final int MAX_BATCH = 1 << 25; //允许切分的最大长度,即batch+BATCH_UNIT是当前切分的长度,这个长度不能大于MAX_BATCH。如果大于,则等于MAX_BATCH; private final Collection<? extends T> collection; //原始数据 private Iterator<? extends T> it; private final int characteristics; // 分割器的特征 private long est; // 数据源的长度(集合的长度) private int batch; // 切分的初始值,创建时一般是0 …… } static final class ArraySpliterator<T> implements Spliterator<T> { private final Object[] array; //原始数据 private int index; // 切分的初始值,创建时一般是0 private final int fence; // 栅栏,即数据源的长度(数组的长度) private final int characteristics;// 分割器的特征 …… //注:这里没有给出切分的增量单位(BATCH_UNIT)和允许切分的最大长度(MAX_BATCH),是因为这种类型的分割器在其 trySplite()方法中,通过index和fence计算出来,自然也就不用给值了。 }

再通过一个示例感受一下Spliterator:

public class StreamTest01 { public static void main(String[] args) { String[] strs = new String[]{"a","b","c","d","e","f","g","h","i","j","k"}; //原始数据 Spliterator<String> spliterator1 = Spliterators.spliterator(strs, 0); //创建分割器 Spliterator<String> spliterator2 = spliterator1.trySplit(); //调用分割方法,切分数据源 spliterator1.forEachRemaining(i -> System.out.print(i+",")); //打印出"分割器1"拥有的所有元素 System.out.println();//用于换行 System.out.println("------------------"); spliterator2.forEachRemaining(i -> System.out.print(i+","));//打印出"分割器2"拥有的所有元素 } } //输出: //f,g,h,i,j,k, //------------------ //a,b,c,d,e public class StreamTest01 { public static void main(String[] args) { String[] strs = new String[]{"a","b","c","d","e","f","g","h","i","j","k"}; //原始数据 Spliterator<String> spliterator1 = Spliterators.spliterator(strs, 0); //创建分割器 spliterator1.tryAdvance(i -> System.out.print(i+",")); //取出a spliterator1.tryAdvance(i -> System.out.print(i+",")); //取出b spliterator1.tryAdvance(i -> System.out.print(i+",")); //取出c spliterator1.tryAdvance(i -> System.out.print(i+",")); //取出d } } //输出:a,b,c,d,

接下来看看IteratorSpliterator的源代码实现(其他两个实现的思想是一样的,只是代码的写法上有些微的差异):

static class IteratorSpliterator<T> implements Spliterator<T> { /** * * 这个方法的主要作用是,在每次调用时,根据事先设置好的切分单位和相关属性 * 将原始数据源进行切分为多个数据源,并将切分出来的数据源包装为新的Spliterator */ @Override public Spliterator<T> trySplit() { Iterator<? extends T> i; long s; if ((i = it) == null) { i = it = collection.iterator(); s = est = (long) collection.size(); } else s = est; if (s > 1 && i.hasNext()) { //判断是否还有元素可以进行拆分 int n = batch + BATCH_UNIT; //获取切分长度 if (n > s) //如果切分长度大于元素的个数,则切分长度等于元素个数 n = (int) s; if (n > MAX_BATCH) //如果切分长度大于最大允许切分长度,则切分长度等于最大可切分长度 n = MAX_BATCH; Object[] a = new Object[n]; //从这一步开始主要做两件事:1、将元素取出,2、将元素包装为Spliterator int j = 0; do { a[j] = i.next(); } while (++j < n && i.hasNext()); batch = j; if (est != Long.MAX_VALUE) est -= j; return new ArraySpliterator<>(a, 0, j, characteristics); } return null; } /** * 这个方法是遍历出所有的元素,因为是集合类型数据,所以使用了集合的迭代器 * 在Spliterator接口中,这个方法会调用tryAdvance()方法,依次取出元素。如: * default void forEachRemaining(Consumer<? super T> action) {do { } while (tryAdvance(action));} * 但仅仅只是一种取出元素的标准,如果实现类可以有自己的取出元素的方式,则也可以直接覆盖 * @param action */ @Override public void forEachRemaining(Consumer<? super T> action) { if (action == null) throw new NullPointerException(); Iterator<? extends T> i; if ((i = it) == null) { i = it = collection.iterator(); est = (long)collection.size(); } i.forEachRemaining(action); } /** * 这个一个取出元素的方法,每次取出一个元素后,Spliterator将不再拥有这个元素 * (可能删除了元素,也可能并没有删除,只是Spliterator再也取不到这个元素了) * @param action * @return */ @Override public boolean tryAdvance(Consumer<? super T> action) { if (action == null) throw new NullPointerException(); if (it == null) { it = collection.iterator(); est = (long) collection.size(); } if (it.hasNext()) { action.accept(it.next()); return true; } return false; } /** * 获取源数据的长度。这个是一个估计值,因为当计算获取数据源的成本很高时,同时又进行拆分; * 此时获取数据源的长度,就有可能导致计算出的数据源长度不精确。或者数据源的长度没有限制,或者是 * 未知的数据源,则无法获取精确地值。 * 在没有进行分组遍历,且Spliterator具有SIZE特性;此时这个方法获取的值一定等于Spliterator遍历 * 所有元素的次数 * @return */ @Override public long estimateSize() { if (it == null) { it = collection.iterator(); return est = (long)collection.size(); } return est; } @Override public int characteristics() { return characteristics; } /** * 获取元素的比较器,由子类自定义,也可以直接抛异常 * @return */ @Override public Comparator<? super T> getComparator() { if (hasCharacteristics(Spliterator.SORTED)) return null; throw new IllegalStateException(); } }

通过上面的描述,Stream的数据源是通过一个Spliterator的实例持有一个原始数据源来提供的。
      Stream的中间操作和终止操作

      在上面的内容中,说到流有三个部分:开端、终端、中端。根据上面的描述,流的开端保存了数据的源,提供了一定的数据源的操作方法。而对于中端和终端,则为进行任何的描述。在JDK的文档中,将流定义为一系列操作的集合,同时JDK文档又将这些操作分为两类,即中间操作和终止操作。看一段代码,直观感受一下中间操作和终止操作:

import java.util.Arrays;
import java.util.List;
import java.util.stream.Stream;

public class StreamTest01 {
	
	public static void main(String[] args) {
		  List<Integer> lis = Arrays.asList(1,2,3,4,5,6,7,8,9);
		  Stream<Integer> stream1 = lis.stream().map(i -> i+1);  //中间操作
		  Stream<Integer> stream2 = stream1.filter(i -> i > 0);  //中间操作
		  Stream<Integer> stream3 = stream2.limit(10);  //中间操作
		  Stream<Integer> stream4 = stream3.distinct(); //中间操作
		  
		  //Optional<Integer> result1 = stream4.findAny(); //终止操作
		  //Optional<Integer> result2 = stream4.findFirst(); //终止操作,两次调用终止操作,会抛异常,流只能被使用一次,可以中间操作,也可以是终止操作
		  stream4.forEach(i -> System.out.print(i));  //终止操作,没有返回值
		  
		  //上面的所有操作有另外的一种写法
		  lis.stream().map(i -> i+1).filter(i -> i > 0)
		  	 .limit(10).distinct().forEach(i -> System.out.print(i));
	} 
}

从上面的代码可以看到,中间操作返回的都是Stream对象;终止操作可能有返回值,可能没有;而且返回的值类型不定。

      接下来从源代码角度看一下中间操作和终止操作都是如何运行的。这里选取map方法和foreach方法作为分析对象,其他方法的实现方式类似。 

//类名:ReferencePipeline;方法:map
public final <R> Stream<R> map(Function<? super P_OUT, ? extends R> mapper) {
	Objects.requireNonNull(mapper);  留意StatelessOp
	return new StatelessOp<P_OUT, R>(this, StreamShape.REFERENCE,  
								 StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
		@Override
		Sink<P_OUT> opWrapSink(int flags, Sink<R> sink) {
			return new Sink.ChainedReference<P_OUT, R>(sink) {
				@Override
				public void accept(P_OUT u) {
					downstream.accept(mapper.apply(u));
				}
			};
		}
	};
}

//类名:ReferencePipeline;方法:forEach
public void forEach(Consumer<? super P_OUT> action) {
	evaluate(ForEachOps.makeRef(action, false));  //调用了AbstractPipeline的evaluate方法
}

//类名:AbstractPipeline;方法:evaluate
final <R> R evaluate(TerminalOp<E_OUT, R> terminalOp) {  //留意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()));
}

在map方法中创建了StatelessOp对象。在foreach方法中通过ForEachOps.makeRef(action, false)方法创建了一个TerminalOp对象。

     StatelessOp比较简单,StatelessOp继承了AbstractPipeline,并覆写了opWrapSink方法。在opWrapSink方法中创建了一个Sink.ChainedReference对象。StatelessOp继承了AbstractPipeline,这里调用StateleddOp的构造方法,实际上是调用了AbstractPipeline的创建中间操作对象的构造方法(前面提到过AbstractPipeline的构造方法有两个,一个构造流的开端,一个构造中间操作),构造了一个AbstractPipeline对象,这个对象表示流的中间操作。

        TerminalOp对象的创建则相对复杂一些。在这之前需要了解两个接口:TerminalOp和TerminalSink,源代码如下:

//接口:TerminalOp
interface TerminalOp<E_IN, R> {
    
    default StreamShape inputShape() { return StreamShape.REFERENCE; }

     
    default int getOpFlags() { return 0; }

     
    default <P_IN> R evaluateParallel(PipelineHelper<E_IN> helper,
                                      Spliterator<P_IN> spliterator) {
        if (Tripwire.ENABLED)
            Tripwire.trip(getClass(), "{0} triggering TerminalOp.evaluateParallel serial default");
        return evaluateSequential(helper, spliterator);
    }

    
    <P_IN> R evaluateSequential(PipelineHelper<E_IN> helper,
                                Spliterator<P_IN> spliterator);
}

//接口:TerminalSink
interface TerminalSink<T, R> extends Sink<T>, Supplier<R> { }

在TerminalOp接口中分别定义了顺序执行的方法(evaluateSequential)和并行执行的方法(evaluateParallel);留心一下会发现,并行计算的方法调用了顺序执行的方法。也就是说,默认情况下是没有并行的(因为并行调用了顺序执行的方法);具体的并行的执行方式需要具体的子类去实现。而对于
TerminalSink接口,则没有任何的实现,直接继承了Sink接口。

接下来再看看ForEachOps.makeRef(action, false)的执行内容:

//类名:ForEachOps ; 方法名:makeRef
public static <T> TerminalOp<T, Void> makeRef(Consumer<? super T> action,boolean ordered) {
	Objects.requireNonNull(action);
	return new ForEachOp.OfRef<>(action, ordered);
}

//类名:ForEachOps.OfRef 
static final class OfRef<T> extends ForEachOp<T> {
	final Consumer<? super T> consumer;

	OfRef(Consumer<? super T> consumer, boolean ordered) {
		super(ordered);
		this.consumer = consumer;
	}

	@Override
	public void accept(T t) {
		consumer.accept(t);
	}
}

//类名:ForEachOp
static abstract class ForEachOp<T> implements TerminalOp<T, Void>, TerminalSink<T, Void> {
	private final boolean ordered;

	protected ForEachOp(boolean ordered) {
		this.ordered = ordered;
	}
	……
}

从上面的代码可以看到ForEachOps.makeRef(action, false)创建了一个 ForEachOps.OfRef 对象。ForEachOps实现了TerminalOp和TerminalSink接口,同时也实现了Sink接口。

无论是StatelessOp还是TerminalOp,最终都关联了Sink接口。其实流的所有操作都是由Sink接口的各种实现类来串联的。看一下Sink接口的代码: 

interface Sink<T> extends Consumer<T> {
    
    default void begin(long size) {}
  
    default void end() {}
  
    default boolean cancellationRequested() {
        return false;
    }
   
    default void accept(int value) {
        throw new IllegalStateException("called wrong accept method");
    }
   
    default void accept(long value) {
        throw new IllegalStateException("called wrong accept method");
    }
  
    default void accept(double value) {
        throw new IllegalStateException("called wrong accept method");
    }
	
	static abstract class ChainedReference<T, E_OUT> implements Sink<T> {
        protected final Sink<? super E_OUT> downstream;

        public ChainedReference(Sink<? super E_OUT> downstream) {
            this.downstream = Objects.requireNonNull(downstream);
        }

        @Override
        public void begin(long size) {
            downstream.begin(size);
        }

        @Override
        public void end() {
            downstream.end();
        }

        @Override
        public boolean cancellationRequested() {
            return downstream.cancellationRequested();
        }
    }
 
    static abstract class ChainedInt<E_OUT> implements Sink.OfInt {
        protected final Sink<? super E_OUT> downstream;

        public ChainedInt(Sink<? super E_OUT> downstream) {
            this.downstream = Objects.requireNonNull(downstream);
        }

        @Override
        public void begin(long size) {
            downstream.begin(size);
        }

        @Override
        public void end() {
            downstream.end();
        }

        @Override
        public boolean cancellationRequested() {
            return downstream.cancellationRequested();
        }
    }
 
    static abstract class ChainedLong<E_OUT> implements Sink.OfLong {
        protected final Sink<? super E_OUT> downstream;

        public ChainedLong(Sink<? super E_OUT> downstream) {
            this.downstream = Objects.requireNonNull(downstream);
        }

        @Override
        public void begin(long size) {
            downstream.begin(size);
        }

        @Override
        public void end() {
            downstream.end();
        }

        @Override
        public boolean cancellationRequested() {
            return downstream.cancellationRequested();
        }
    }
 
    static abstract class ChainedDouble<E_OUT> implements Sink.OfDouble {
        protected final Sink<? super E_OUT> downstream;

        public ChainedDouble(Sink<? super E_OUT> downstream) {
            this.downstream = Objects.requireNonNull(downstream);
        }

        @Override
        public void begin(long size) {
            downstream.begin(size);
        }

        @Override
        public void end() {
            downstream.end();
        }

        @Override
        public boolean cancellationRequested() {
            return downstream.cancellationRequested();
        }
    }	
}

        这里采用了模板方法模式,规定了Sink方法的调用必须是begin,accept,end的顺序,可以重复调用,但必须保证调用方法的顺序(begin,accept,end)。

        在这个接口中有三个重载的accept方法,却没有accept(Object  value),这是因为Sink接口继承了Consumer接口,在Consumer接口中已经有一个接收Object参数的accept方法了。

     此外,Sink接口中定义了四个实现类(ChainedReference,ChainedInt,ChainedLong,ChainedDouble),这四个实现类中都分别持有一个downstream的Sink对象,这个对象是实现Stream中操作链接的关键。

       前面都是准备阶段,接下来看看Stream是如何将所有的操作链接起来进行的。在调用map方法的时候,留心的话,你就会发现,map方法仅仅只是创建了一个StatelessOp对象,却没有调用任何的执行代码。这其实就是流的惰性计算。任何的中间操作都没有执行任何的计算代码,只有在遇到终止操作时才会触发流的执行。接下来再看看终止操作是如何触发流的执行的,源代码:

//类名:ReferencePipeline;方法名:forEach
public void forEach(Consumer<? super P_OUT> action) {
	evaluate(ForEachOps.makeRef(action, false));
}

//类名:AbstractPipeline;方法名:evaluate
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()));
}

//类名:ForEachOp;方法名:evaluateSequential
public <S> Void evaluateSequential(PipelineHelper<T> helper,
                                           Spliterator<S> spliterator) {
	return helper.wrapAndCopyInto(this, spliterator).get();
}

//类名:AbstractPipeline;方法名:wrapAndCopyInto。(AbstractPipeline继承了PipelineHelper)
final <P_IN, S extends Sink<E_OUT>> S wrapAndCopyInto(S sink, Spliterator<P_IN> spliterator) {
	copyInto(wrapSink(Objects.requireNonNull(sink)), spliterator); //这里的copyInto方法触发了执行
	return sink;
}

//类名:AbstractPipeline;方法名:wrapSink
final <P_IN> Sink<P_IN> wrapSink(Sink<E_OUT> sink) { //这个方法是关键,操作就是在这里进行连接的
	Objects.requireNonNull(sink);

	for ( @SuppressWarnings("rawtypes") AbstractPipeline p=AbstractPipeline.this; p.depth > 0; p=p.previousStage) {
		sink = p.opWrapSink(p.previousStage.combinedFlags, sink);
	}
	return (Sink<P_IN>) sink;
}

在wrapSink中,有一个参数需要解释一下——p.depth:这个参数表示Stream的深度,即有多少个操作需要链接。我们在创建 AbstractPipeline对象时,每进行一次中间操作对象的创建,这个属性值就会加一。如下:

AbstractPipeline(AbstractPipeline<?, E_IN, ?> previousStage, int opFlags) {
        if (previousStage.linkedOrConsumed)
            throw new IllegalStateException(MSG_STREAM_LINKED);
        previousStage.linkedOrConsumed = true;
        previousStage.nextStage = this;

        this.previousStage = previousStage; //前一个操作
        this.sourceOrOpFlags = opFlags & StreamOpFlag.OP_MASK;
        this.combinedFlags = StreamOpFlag.combineOpFlags(opFlags, previousStage.combinedFlags);
        this.sourceStage = previousStage.sourceStage;
        if (opIsStateful())
            sourceStage.sourceAnyStateful = true;
        this.depth = previousStage.depth + 1; //每构造一次,就会加一
    }

在wrapSink方法中再次调用了opWrapSink,这个方法就是前面在调用Stream的map方法时,重写的方法。如下:

public final <R> Stream<R> map(Function<? super P_OUT, ? extends R> mapper) {
        Objects.requireNonNull(mapper);
        return new StatelessOp<P_OUT, R>(this, StreamShape.REFERENCE,
                                     StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
            @Override
            Sink<P_OUT> opWrapSink(int flags, Sink<R> sink) { //重写方法
                return new Sink.ChainedReference<P_OUT, R>(sink) {
                    @Override
                    public void accept(P_OUT u) {  //重写了accept方法,在方法中调用downstream的accept方法
                        downstream.accept(mapper.apply(u)); //经过这里,就可以将所有的操作连接起来
                    }
                };
            }
        };
    }

接下里再看看是如何触发这些操作的,触发这些操作的代码在copyInto方法中。代码如下:

 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); //流有短路特性,则执行
	}
}

当流没有短路特性时,调用了spliterator的forEachRemaining方法,传入了刚才包装的那个sink对象。spliteator的实现类很多,执行形式也有区别,但是在spliterator中都会调用wrapsink的accept方法。这个wrapsink的accept方法中调用downstream.accept方法,这个时候就会触发这个流的所有操作的执行。





版权声明:本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系管理员进行删除。
喜欢 (0)