由浅入深聊清楚 Stream 流原理

本文已参与好文召集令活动，点击查看：后端、大前端双赛道投稿，2万元奖池等你挑战！

首先分享之前的所有文章 , 欢迎点赞收藏转发三连下次一定 >>>> 😜😜😜

文章合集 : 🎁 juejin.cn/post/694164…

Github : 👉 github.com/black-ant

CASE 备份 : 👉 gitee.com/antblack/ca…

一 . 前言

这一篇从函数式接口一步步说起 , 来聊一聊 Java 8 中这个特性的原理以及优点

函数式编程之前一直是基于使用 , 最近梳理源码的时候 ,发现这个概念真的无处不在 , 索性把这一块完整的梳理处理

Java Stream 的基本使用可以看这一篇 :操作手册 : Stream 流处理手册

本篇文章会分为 2个主体 :

• Java 的函数式编程
• Java 的 Stream 原理

二 . 函数式编程原理

函数式编程主要有四种接口 : Consumer、Supplier、Predicate、Function , 每个函数接口都有一个单独的抽象方法

注解: 函数式接口通过 @FunctionalInterface 注解进行标注 , 该注解只能标注在 有且只有一个抽象方法 的接口. (PS:函数式接口不一定非要加该注解)

java复制代码// 以 Consumer 为例 , 可以看到 , 这里的唯一抽象方法是 accept 
@FunctionalInterface
public interface Consumer<T> {

    void accept(T t);

    default Consumer<T> andThen(Consumer<? super T> after) {
        Objects.requireNonNull(after);
        return (T t) -> { accept(t); after.accept(t); };
    }
}

这里有几个概念 :

1. 函数式编程接口中只能有一个抽象方法
2. 可以有 static 和 default 方法 (PS : 不属于抽象方法)
3. 可以重写 Object 方法 (PS : 函数本身继承于 Object , 这里后面会执行看看)
4. 注解非必须

2.1 函数式编程的使用

先来自定义一个函数式编程的流程 , 了解一下其细节 :

2.1.1 箭头函数Demo

java复制代码// 来看一下函数式编程的简单使用 :

// Step 1 : 自定义函数式接口
@FunctionalInterface
public interface FunctionInterface<T, D> {

    /**
     * 函数接口抽象方法 , 最终执行方法
     * PS : 至此泛型
     */
    T invoke(D input);

    /**
     * Object 中方法 ,该方法不违背函数式编程原则
     *
     * @param var1
     * @return
     */
    boolean equals(Object var1);

    /**
     * default不是抽象方法 , 不违背原则
     */
    default void defaultMethod() {
    }

    /**
     * static不是抽象方法
     */
    static void staticMethod(String msg) {
        System.out.println(msg);
    }

}


// Step 2 : 定义调用函数
public void testFunction() {
    logger.info("------> [执行函数式方法] <-------");

    // 方法一 : 传入代码块
    invokeInterface((input) -> {
        return input + "-output1";
    });

    // 方法二 : 直接访问
    invokeInterface((input) -> input + "-output2");
    
    // 方法三 : 传入对象
    FunctionInterface<String, String> funtion = (input) -> input + "-output3";
    invokeInterface(funtion);

}


public void invokeInterface(FunctionInterface<String, String> funtion) {
    String response = funtion.invoke("test");
    logger.info("------> this is output :{} <-------", response);
}

// 打印结果
FunctionService    : ------> this is output :test-output1 <-------
FunctionService    : ------> this is output :test-output2 <-------
FunctionService    : ------> this is output :test-output3 <-------

2.1.2 双冒号函数Demo

java复制代码public void testDoubleFunction() {

    // 案例一 : 传入实例方法
    Consumer consumer = System.out::println;
    invokeDoubleFunction(consumer);

    // 案例二 : 传入静态方法
    Consumer<String> consumer2 = FunctionInterface::staticMethod;
    invokeDoubleFunction(consumer2);

    // 案例三 : 传入超类
    Consumer<String> consumer3 = super::superMethod;
    invokeDoubleFunction(consumer3);

    // 案例四 : 传入构造函数
    Consumer<ArrayList> consumer4 = ArrayList::new;
    invokeDoubleFunction2(consumer4);
    
    // 常用案例 , 传入自定义方法
    Consumer<String> createRoot = (msg) -> {
        logger.error("E----> error :{} -- content :{}", e.getClass(), e.getMessage());
    };
    invokeDoubleFunction2(createRoot);;
}

public void invokeDoubleFunction(Consumer consumer) {
    Stream<String> stream = Stream.of("aaa", "bbb", "ddd", "ccc", "fff");
    stream.forEach(consumer);
}

public void invokeDoubleFunction2(Consumer consumer) {
    Stream<Collection<String>> stream = Stream.of(Arrays.asList("aaa", "bbb"));
    stream.forEach(consumer);
}

2.2 相关注解 / 接口

上面看完了函数式编程的自定义方式 , 这里来看一下相关的接口 👉

Java 已知的函数式接口有四个主要的 : Consumer、Supplier、Predicate、Function , 以及其他类似的 : IntConsumer , IntSupplier , UnaryOperator 等等 , 这里我们只要就 4 种主要的进行一个简单的分析

Consumer : 消费接口
理解 : 该接口函数的目的是为了消费传入的参数 , 主要集中在参数的使用上

结构 : 从结构上就可以看出其特性 ,它是接收一个参数 ,但是没有返回 ( void )

java复制代码public interface Consumer<T> {
    void accept(T t);
}

Supplier : 供给型
理解 : Supplier 的作用场景主要单纯的获取资源

结构 : 没有输入 ,只有返回

java复制代码public interface Supplier<T> {
    T get();
}

Predicate : 断言型(谓词型)

理解 : 断言型表示一个参数的断言(布尔值函数 , PS : 文档里面经常说谓词 ,但是我感觉翻译为断言更符合)

结构 : 传入对象 ,返回布尔

java复制代码public interface Predicate<T> {
    boolean test(T t);
}

// 用法 :
stream.filter(predicate).collect(Collectors.toList());

Function : 功能型
理解 : 接受一个参数并产生一个结果的函数 , 也是功能适用性最好的方式

java复制代码public interface Function<T, R> {
    R apply(T t);
}

// 用法 :
Function<String, Integer> function = new Function<String, Integer>() {
	@Override
	public Integer apply(String s) {
		return s.length();//获取每个字符串的长度，并且返回
	}
};
Stream<Integer> stream1 = stream.map(function);

Consumer 与 IntConsumer 类似接口的区别 , 以及优化思考

可以看到 , 其中每一个函数接口中都为其手动扩展了一些基本类型的接口 , 例如 IntConsumer 等等

之前从资料中了解到 , 这样的目的是为了对基本类型进行优化 , 但是看源码的时候并没有直接的区别 :

个人考虑了一下 , 泛型主要是编译器进行处理 , 在实际的使用阶段是没什么影响的 , 而 Java 基本类型的包装功能实际上也没什么优化作用 , 那么这里到底优化了什么呢 ?还是说单纯的为了更清晰 ?

// PS : 我实在不相信 JDK 里面会做这种事 , 所以 , 一定还有位置!!!

联想一下 , 在编译器之前就确定实际类型 , 那么一定在业务代码中有直观的地方去处理这种类型 , 而避免使用反射等方式在运行时处理类型.

TODO : 挺有意思的 , 不过以后再看看 …

2.2 方法函数原理

前面说了函数式编程的用法 , 现在进入正题 , 看一看函数式编程的原理 :

箭头函数的使用 :

java复制代码// 从自定义的案例来分析 , 传入的实际上是一个接口对象

方法函数是 lambda 的概念 , 其原理是 Class 层面的使用方式,  需要反编译看一下操作 : 

// TODO : 反编译暂无必要 , 后续完善

三 . Steam 深入

3.1 Stream 体系结构

从图里面可以看到 , 基本上体系得结构都是一致的 , 有点像树状结构 :

第一层是基础抽象类 : BaseStream

第二层是抽象层次 , 包含5个只要类 : AbstractPipeline / Stream / IntStream / DoubleStream / LongStream

第三层是实现主类 : DoublePipeline / LongPipeline / IntPipeline / ReferencePipeline

第四层为内部类 , 每个实现都有对应的几个 : Head / StatelessOp / StatefulOp / OfInt

3.2 Stream 运行原理

可以看到 , Stream 结构中 , 主要基于 Pipeline 的概念 , 其中额外对三个基本类型做了优化 .

同时通过 StatefulOp、StatelessOp用于对应有状态和无状态中间操作 , 做一个简单的概念整理 :

流程图解 (一图带你了解主流程)

---

前置补充 : AbstractPipeline 类

AbstractPipeline 的作用 :

stage 是一种虚拟概念 , AbstractPipeline表示流管道的初始部分，封装了流源和零个或多个中间操作 , 一个 AbstractPipeline 被看成一个 stage , 其中每个阶段要么描述流源，要么描述中间操作.

stage 属性 :

AbstractPipeline 中有三个 stage 概念 (用于标注空间结构)

F- AbstractPipeline sourceStage : 指向管道链的头部(如果这是源阶段，则为self)

F- AbstractPipeline previousStage : “上游” 管道，如果这是源级，则为空

F- AbstractPipeline nextStage : 管道中的下一个阶段，如果这是最后一个阶段，则为空

上面三个属性是用于标注空间结构 , 意味着流程走到了哪里 , 剩下的就是如何标注行为

行为类型 :

Head 、StatefulOp 、StatelessOp , 这三个属性都继承了 AbstractPipeline , 同时他们标识了三种操作类型

Head : 表示第一个Stage，也就是source stage , 主要是资源收集

StatefulOp : 有状态操作

StatelessOp : 无状态操作

通常执行逻辑 : Head -> StatefulOp -> StatelessOp

其他属性 :

java复制代码// 上游管道，第一次创建流则为null
previousStage = null; 
// 源切割器 , 在每个工具类中自行实现
sourceSpliterator = source; 
//此管道对象中表示的中间操作的操作标志。
sourceOrOpFlags = sourceFlags & StreamOpFlag.STREAM_MASK; 

// 源和该管道对象所表示的操作的所有操作的合并源和操作标志
combinedFlags = (~(sourceOrOpFlags << 1)) & StreamOpFlag.INITIAL_OPS_VALUE;

/**
* 非并行 : 与源流的深度(步数) 
* 并行 : 表示前面的有状态操作数
**/
depth = 0; 

/**
* 管道是否为并行
**/
parallel = parallel;

AbstractPipeline 构造函数

java复制代码AbstractPipeline(Supplier<? extends Spliterator<?>> source,
                     int sourceFlags, boolean parallel) {
	this.previousStage = null;  // 上游 stage
	this.sourceSupplier = source;
	this.sourceStage = this;
	this.sourceOrOpFlags = sourceFlags & StreamOpFlag.STREAM_MASK;
	this.combinedFlags = (~(sourceOrOpFlags << 1)) & StreamOpFlag.INITIAL_OPS_VALUE;
	this.depth = 0;
	this.parallel = parallel;
}

3.2.1 Stream 的流程

前置要点

这里梳理了一下从相关博客中了解到的 Stream 操作的全部要点 , 便于后文学习 :

java复制代码// 主流程
- 每一个操作会创建一个 stage
- 上游 sink 通过 Sink AbstractPipeline.opWrapSink 方法查找下游 Sink

// 结束流程
- 结束操作不会创建新的流水线阶段(Stage)
- 结束操作会包装一个自己操作的 Sink , 自己作为流水线的最后一个 sink 
    
// 收集流程
- 对于 Boolean 和 Optional 操作 ,会在对应 Sink 中记录    
- 对于归约操作，最终结果放在用户调用时指定的容器中    
- 返回数组会先放在一个 Node 数据结构中 
    
// 并行操作 
-  并行通过 ForkJoin 执行 Task

使用案例

java复制代码// 按照以下案例来看一下主要流程 :
List<String> randomValues = Arrays.asList(
	"E11", "D12", "A13", "F14", "C15", "A16",
	"B11", "B12", "C13", "B14", "B15", "B16",
	"F12", "E13", "C11", "C14", "A15", "C16",
	"F11", "C12", "D13", "E14", "D15", "D16"
);

randomValues.stream().filter(value -> value.startsWith("C")).sorted().forEach(System.out::println);

3.2.1.1 Step 1 : Stream 的创建

通常可以通过 Collection 或者 Array 实现 Stream 的创建 , 这里我们不关注太多的细节 , 只是看一下创建出来的是什么样的

1. 通过 Spliterator 切割集合 (集合或者数组本身的方法)
2. 通过 StreamSupport.stream 构建一个 Stream

java复制代码// 可以看到 , 通过 ReferencePipeline.Head 构建了一个 Stream
return new ReferencePipeline.Head<>(spliterator,StreamOpFlag.fromCharacteristics(spliterator),parallel);

// 补充 spliterator :
Spliterator 有很多实现类 , 因为之前使用的是 ArrayList , 所以以ArrayListSpliterator 为例 , 其中主要有三个属性 :
private final ArrayList<E> list; // 传入当前 Stream 的集合
private int index; // 当前的索引 , 在增长和切割的时候修改
private int fence; // 当前集合数量 ,当为 -1 时标识已经全部结束
private int expectedModCount; // 当 fence 设置时初始化


// 补充 StreamOpFlag : 该对象把 Spliterator 转换为了一个流标志 , 用来标识流的特征 , 特征常见的有以下几种 :
public static final int ORDERED    = 0x00000010; // 定义相遇顺序
public static final int DISTINCT   = 0x00000001; // 定义唯一特征
public static final int SORTED     = 0x00000004; // 遵循已定义的排序顺序的特征值 , 存在可比性
public static final int SIZED      = 0x00000040; // 表示完整遍历将遇到的元素数量的精确计数
public static final int NONNULL    = 0x00000100; // 表示源保证遇到的元素不为空的特征值
public static final int IMMUTABLE  = 0x00000400; // 元素源不能被结构修改的特征值 , 即不可修改
public static final int CONCURRENT = 0x00001000; // 表示元素源可以被多线程安全地并发修改(允许添加、替换和/或删除)，而不需要外部同步的特征值
public static final int SUBSIZED = 0x00004000; // 所有子spliterator，无论是直接的还是间接的，都将被计数 (子类计数)



// parallel : 返回的流是否为并行的

3.2.1.2 Step 2 : Filter 过滤

前面看了 Stream 的创建流程 :

java复制代码public final Stream<P_OUT> filter(Predicate<? super P_OUT> predicate) {
    Objects.requireNonNull(predicate);
    return new StatelessOp<P_OUT, P_OUT>(this, StreamShape.REFERENCE,
                                 StreamOpFlag.NOT_SIZED) {
        @Override
        Sink<P_OUT> opWrapSink(int flags, Sink<P_OUT> sink) {
            return new Sink.ChainedReference<P_OUT, P_OUT>(sink) {
                @Override
                public void begin(long size) {
                    downstream.begin(-1);
                }

                @Override
                public void accept(P_OUT u) {
                    // 通过传入的方法函数进行校验
                    if (predicate.test(u))
                        // 校验成功加入输出流 (PS : 此处会讲其加入后文的 List )
                        downstream.accept(u);
                }
            };
        }
    };
}

// 补充 : StreamShape 是 描述流抽象的类型特征的枚举 , 其中有四种属性
REFERENCE : object
INT_VALUE 
LONG_VALUE
DOUBLE_VALUE

这一段代码里面集中说了什么 :

1. 构建了一个 StatelessOp
2. Sink.ChainedReference 是一个反向回调逻辑 ,在流 foreach 的时候 , 这个操作才会执行

补充 : 核心操作 wrapSink

java复制代码// 作用 : 
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;
}

着重分析一下这个环节 :

什么是管道 , 管道就是一个队列 , 一头进水 , 一头才能出水

而filter 相当于其中的分流阀门 , 把一部分水分出去 , 但是一切的前提就是 , 管道的开关要打开

也就是说当流程走到 foreach 等 Terminal 操作的时候 , 流才开始运行 , 其中设置的中间操作才会执行

3.2.1.3 Step 3 : forEach 流程

forEach 入口

java复制代码C- ReferencePipeline (java.util.stream.SortedOps$OfRef)
public void forEach(Consumer<? super P_OUT> action) {
	evaluate(ForEachOps.makeRef(action, false));
}

// --> evaluate 具体实现 , 执行管道处理
final <R> R evaluate(TerminalOp<E_OUT, R> terminalOp) {
    assert getOutputShape() == terminalOp.inputShape();
    // linkedOrConsumed -> 此管道已被应用或使用 , 意味着管道只能单次使用
    if (linkedOrConsumed)
        throw new IllegalStateException(MSG_STREAM_LINKED);
    linkedOrConsumed = true;
	
    // 是否为并行流
    return isParallel()
           ? terminalOp.evaluateParallel(this, sourceSpliterator(terminalOp.getOpFlags()))
           : terminalOp.evaluateSequential(this, sourceSpliterator(terminalOp.getOpFlags()));
}


// 补充 : Parallel 的对象区别
evaluateParallel : 使用指定的PipelineHelper对操作执行并行计算
evaluateSequential : 使用指定的参数执行操作的顺序计算

内部流程第一步 :　wrapSink 构建 Sink

java复制代码public <S> Void evaluateSequential(PipelineHelper<T> helper,Spliterator<S> spliterator) {
	return helper.wrapAndCopyInto(this, spliterator).get();
}

@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;
}

// 补充  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;
}

public Sink<T> opWrapSink(int flags, Sink<T> sink) {
	Objects.requireNonNull(sink);

	// If the input is already naturally sorted and this operation
	// also naturally sorted then this is a no-op
	if (StreamOpFlag.SORTED.isKnown(flags) && isNaturalSort)
		return sink;
	else if (StreamOpFlag.SIZED.isKnown(flags))
		return new SizedRefSortingSink<>(sink, comparator);
	else
		return new RefSortingSink<>(sink, comparator);
}


@Override
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);
        // 在所有数据已经发送后，必须调用 end 结束 
        wrappedSink.end();
        
        // PS : end 结束后不能再次调用 begin 和 accept
    }
    else {
        copyIntoWithCancel(wrappedSink, spliterator);
    }
}

内部流程第二步 : 发起 Foreach 循环

JAVA复制代码public void end() {
    // 此时的 list 已经过滤完成 ,  此处进行排序
	list.sort(comparator);
    
    // 和上文类似 , downstream 同样是一个 Sink
	downstream.begin(list.size());
	if (!cancellationWasRequested) {
         // PS : 此处的list 前文 Filter 中成功添加
		list.forEach(downstream::accept);
	} else {
		for (T t : list) {
			if (downstream.cancellationRequested()) break;
				downstream.accept(t);
		}
	}
	downstream.end();
	list = null;
}

内部流程第三步 : foreach 主流程

JAVA复制代码public void forEach(Consumer<? super E> action) {
    Objects.requireNonNull(action);
    final int expectedModCount = modCount;

    final E[] elementData = (E[]) this.elementData;
    final int size = this.size;
    for (int i=0; modCount == expectedModCount && i < size; i++) {
        action.accept(elementData[i]);
    }
    if (modCount != expectedModCount) {
        throw new ConcurrentModificationException();
    }
}

内部流程第四步 : 执行 accept

JAVA复制代码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) {
        // 此处执行 foreach 中传入的方法函数
        consumer.accept(t);
    }
}

3.2.1.4 补充 Sink :

作用 :

• begin(long size) : 开始遍历元素之前调用该方法，通知Sink做好准备
• end() : 所有元素遍历完成之后调用，通知Sink没有更多的元素了
• cancellationRequested() : 是否可以结束操作，可以让短路操作尽早结束 (短路操作必须实现)
• accept(T t) : 遍历元素时调用，接受一个待处理元素，并对元素进行处理 (PS : 链表中的 Stage 通过 accept 下层调用)

java复制代码// 调用流程 : 
每个Stage都会将自己的操作封装到一个Sink里，后一个Stage只需调用前一个Stage的accept()方法 , 
有点类似于递归的思想 ,但是又有不同
    
递归是由底层像顶层 , Stream 是执行底层后 , 从顶层一层层向下

四 . 要点补充

4.1 Map 流程

JAVA复制代码// 
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) {
                    // u 即为当前对象 , mapper.apply 用于将此函数应用于给定的参数
                    downstream.accept(mapper.apply(u));
                }
            };
        }
    };
}

4.2 Collection 流程

java复制代码public final <R, A> R collect(Collector<? super P_OUT, A, R> collector) {
    A container;
    if (isParallel()
            && (collector.characteristics().contains(Collector.Characteristics.CONCURRENT))
            && (!isOrdered() || collector.characteristics().contains(Collector.Characteristics.UNORDERED))) {
        container = collector.supplier().get();
        BiConsumer<A, ? super P_OUT> accumulator = collector.accumulator();
        forEach(u -> accumulator.accept(container, u));
    }
    else {
        container = evaluate(ReduceOps.makeRef(collector));
    }
    return collector.characteristics().contains(Collector.Characteristics.IDENTITY_FINISH)
           ? (R) container
           : collector.finisher().apply(container);
}


// 此处会注入 Supplier 对象
public static <T> Collector<T, ?, List<T>> toList() {
	return new CollectorImpl<>((Supplier<List<T>>) ArrayList::new, List::add,
                                   (left, right) -> { left.addAll(right); return left; },
                                   CH_ID);
}

// 实际此处
static class CollectorImpl<T, A, R> implements Collector<T, A, R> {
        private final Supplier<A> supplier;
        private final BiConsumer<A, T> accumulator;
        private final BinaryOperator<A> combiner;
        private final Function<A, R> finisher;
        private final Set<Characteristics> characteristics;
    
}

// 执行 ReduceOps 对象的时候会调用 supplier
public static <T, I> TerminalOp<T, I>
    makeRef(Collector<? super T, I, ?> collector) {
        Supplier<I> supplier = Objects.requireNonNull(collector).supplier();
        BiConsumer<I, ? super T> accumulator = collector.accumulator();
        BinaryOperator<I> combiner = collector.combiner();
        class ReducingSink extends Box<I>
                implements AccumulatingSink<T, I, ReducingSink> {
            @Override
            public void begin(long size) {
                state = supplier.get();
            }

            @Override
            public void accept(T t) {
                accumulator.accept(state, t);
            }

            @Override
            public void combine(ReducingSink other) {
                state = combiner.apply(state, other.state);
            }
        }
        return new ReduceOp<T, I, ReducingSink>(StreamShape.REFERENCE) {
            @Override
            public ReducingSink makeSink() {
                return new ReducingSink();
            }

            @Override
            public int getOpFlags() {
                return collector.characteristics().contains(Collector.Characteristics.UNORDERED)
                       ? StreamOpFlag.NOT_ORDERED
                       : 0;
            }
        };
    }

4.3 stateless 和 stateful 处理的区别

• stateless : 无状态操作
• stateful : 有状态操作

java复制代码    abstract static class StatelessOp<E_IN, E_OUT>
            extends ReferencePipeline<E_IN, E_OUT> {

        StatelessOp(AbstractPipeline<?, E_IN, ?> upstream,
                    StreamShape inputShape,
                    int opFlags) {
            super(upstream, opFlags);
            assert upstream.getOutputShape() == inputShape;
        }

        @Override
        final boolean opIsStateful() {
            return false;
        }
    }

    abstract static class StatefulOp<E_IN, E_OUT>
            extends ReferencePipeline<E_IN, E_OUT> {

        StatefulOp(AbstractPipeline<?, E_IN, ?> upstream,
                   StreamShape inputShape,
                   int opFlags) {
            super(upstream, opFlags);
            assert upstream.getOutputShape() == inputShape;
        }

        @Override
        final boolean opIsStateful() {
            return true;
        }

        @Override
        abstract <P_IN> Node<E_OUT> opEvaluateParallel(PipelineHelper<E_OUT> helper,
                                                       Spliterator<P_IN> spliterator,
                                                       IntFunction<E_OUT[]> generator);
    }


// 对比 : 
1 . StatefulOp 额外实现了 opEvaluateParallel

4.4 并行处理

java复制代码// 前面已经了解到创建 evaluate 时 , 会通过 evaluateParallel 进行并行操作
C- AbstractPipeline
final <R> R evaluate(TerminalOp<E_OUT, R> terminalOp) {
    // ... 是否为并行流
    return isParallel()
           ? terminalOp.evaluateParallel(this, sourceSpliterator(terminalOp.getOpFlags()))
           : terminalOp.evaluateSequential(this, sourceSpliterator(terminalOp.getOpFlags()));
}

// 原理 : 
Stream的并行处理是基于ForkJoin框架的 , 每个 Op 中创建的方式不同 ,但是都是创建了一个 Task
    
// C- ReduceOp
public <P_IN> R evaluateParallel(PipelineHelper<T> helper,Spliterator<P_IN> spliterator) {
	return new ReduceTask<>(this, helper, spliterator).invoke().get();
}

// C- MatchOp
public <S> Boolean evaluateParallel(PipelineHelper<T> helper,Spliterator<S> spliterator) {
	return new MatchTask<>(this, helper, spliterator).invoke();
}

// C- ForEachOp
public <S> Void evaluateParallel(PipelineHelper<T> helper,Spliterator<S> spliterator) {
	if (ordered)
		new ForEachOrderedTask<>(helper, spliterator, this).invoke();
	else
		new ForEachTask<>(helper, spliterator, helper.wrapSink(this)).invoke();
	return null;
}

// C- FindOp
public <P_IN> O evaluateParallel(PipelineHelper<T> helper,Spliterator<P_IN> spliterator) {
	return new FindTask<>(this, helper, spliterator).invoke();
}

java复制代码public void compute() {
        Spliterator<P_IN> rs = spliterator, ls; // right, left spliterators
        long sizeEstimate = rs.estimateSize();
        long sizeThreshold = getTargetSize(sizeEstimate);
        boolean forkRight = false;
        @SuppressWarnings("unchecked") K task = (K) this;
        while (sizeEstimate > sizeThreshold && (ls = rs.trySplit()) != null) {
            K leftChild, rightChild, taskToFork;
            task.leftChild  = leftChild = task.makeChild(ls);
            task.rightChild = rightChild = task.makeChild(rs);
            task.setPendingCount(1);
            if (forkRight) {
                forkRight = false;
                rs = ls;
                task = leftChild;
                taskToFork = rightChild;
            }
            else {
                forkRight = true;
                task = rightChild;
                taskToFork = leftChild;
            }
            taskToFork.fork();
            sizeEstimate = rs.estimateSize();
        }
        task.setLocalResult(task.doLeaf());
        task.tryComplete();
}

这里可以看一下这位前辈的文章 , 写的很清楚 , 以下是搬运的图片 @ Java8 Stream原理深度解析 - 知乎 (zhihu.com)

4.5 Op 模块体系

java复制代码// PS : 这里每一个 Op 都会创建一个 Sink

 copyInto(wrapSink(Objects.requireNonNull(sink)), spliterator);
 
 
 // 核心就是 : wrapSink , 创建后最终调用

4.6 多个 stage 处理

• 通过 构造器设置
• depth 用于设置深度

其他结构如图所示

总结

Stream 原理并没有深入太多 , 主要是对其原理有一定的好奇 ,只对一个流程进行了分析.

Stream 的源码读的很爽 , 很少看到结构这么有趣的代码.

此处还没有完全分析清楚 Stream 是如何通过并行去实现高效处理的 , 下一篇我们看一下性能分析

思考

花了这么久的时间 , 梳理完了这些源码 , 总要从里面学到点什么 , 这里试着做一点总结 :

• Stream 的流程非常有趣 , 它是一种类似于递归但是又略有不同的结构 , 底层逻辑是开关 , 用于开启整个流程 , 当水流动的时候 , 还是从头开始执行
• Stream 的结构体系也很有意思 , 有点类型于链表的 Node next and pre , 只不过其中是虚拟的 stage 对象
• 在继承体系上 , 第一感觉是整整齐齐 , 而且很细致 , 在接口构造上 , 很有参考的价值

附录

Stream 常见方法 :

参考

@ 深入理解Java8中Stream的实现原理_lcgoing的博客-CSDN博客_stream原理

@ Java8 Stream原理深度解析 - 知乎 (zhihu.com)

@ www.javabrahman.com/java-8/unde…)

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有