从0开始实现java Stream功能
java8 stream介绍
java8新增了stream流的特性，能够让用户以函数式的方式、更为简单的操纵集合等数据结构，并实现了用户无感知的并行计算。
相信很多人在使用过java8的streamAPI接口之后，都会对其实现原理感到好奇，但往往在看到jdk的stream源码后却被其复杂的抽象、封装给弄糊涂了，而无法很好的理解其背后的原理。究其原因，是因为jdk的stream源码是高度工程化的代码，工程化的代码为了效率和满足各式各样的需求，会将代码实现的极其复杂，不易理解。
在这里，我们将抛开jdk的实现思路，从零开始实现一个stream流。
我们的stream流同样拥有惰性求值，函数式编程接口等特性，并兼容jdk的Collection等数据结构(但不支持并行计算 orz)。
相信在亲手实现一个stream流的框架之后，大家能更好的理解流计算的原理。
stream的优点
在探讨探究stream的实现原理和动手实现之前，我们先要体会stream流计算的独特之处。
举个例子： 有一个List列表,我们需要获得年龄为70岁的前10个Person的姓名。
**过程式的解决方案：**稍加思考，我们很快就写出了一个过程式的解决方案(伪代码)：
List personList = queryDB(); // 获得List
int limit = 10; // 限制条件
List nameList = new ArrayList(); // 收集的姓名集合
for(Person personItem : personList){if(personItem.age == 70){ // 满足条件nameList.add(personItem.name); // 加入姓名集合if(nameList.size() >= 10){ // 判断是否超过限制break;}}
}
return nameList;
**函数式stream解决方案：**下面我们给出一种基于stream流的解决方案(伪代码)：
List personList = queryDB(); // 获得List
List nameList = personList.stream().filter(item->item.age == 70) // 过滤条件.limit(10) // limit限制条件.map(item->item.name) // 获得姓名.collect(Collector.toList()); // 转化为list
return nameList;
两种方案的不同之处：
从函数式的角度上看，过程式的代码实现将收集元素、循环迭代、各种逻辑判断耦合在一起，暴露了太多细节。当未来需求变动和变得更加复杂的情况下，过程式的代码将变得难以理解和维护(需要控制台打印出 年龄为70岁的前10个Person中，姓王的Person的名称）。
函数式的解决方案解开了代码细节和业务逻辑的耦合，类似于sql语句，表达的是**“要做什么"而不是"如何去做”**，使程序员可以更加专注于业务逻辑，写出易于理解和维护的代码。
List personList = queryDB(); // 获得List
personList.stream().filter(item->item.age == 70) // 过滤条件.limit(10) // limit限制条件.filter(item->item.name.startWith("王")) // 过滤条件.map(item->item.name) // 获得姓名.forEach(System.out::println);
自己实现stream流功能
stream流在使用过程中的三个阶段：
-
生成并构造一个流 (List.stream 等方法)
-
在流的处理过程中添加或绑定惰性求值流程 (map、flatMap、filter、limit 等方法)
-
对流使用强制求值函数，计算最终结果 (max、collect、forEach等方法
**stream流的特点：**流的结构类似于链表，不同的是流的下一个节点是以表达式的形式进行封装的，我们在调用map，flagMap等操作时并没有对节点进行操作，而操作的是节点所对应的表达式。只有在调用强制求值操作时才会调用表达式获取真正的流对象。(惰性求值)
举个例子：有一个从1到5的流，我们调用map将每个元素乘以10最后进行打印：
//在map之前，每个节点的表达式是这样的 Supplier numSup = ()->num，调用map后其实是对表达式再次进行封装变为
// Supplier newSup= ()->numSup.apply()*10
IntStream.range(1,6).map(n -> n*10) // 元素*10.forEach(System.out::println); // 打印
stream流定义
流本身
@AllArgsConstructor(access = AccessLevel.PROTECTED)
@NoArgsConstructor(access = AccessLevel.PROTECTED)
public class StreamNode {//当前节点的数据protected T value;//用于获取下一个节点的表达式protected StreamNodeSupplier nextNode;//是否开始节点protected boolean start;//是否结束节点public boolean isEnd(){return Objects.isNull(this.nextNode);}}
StreamNodeSupplier
public class StreamNodeSupplier {public Supplier> supplier;public StreamNodeSupplier(Supplier> supplier){this.supplier = supplier;}public StreamNode get(){return this.supplier.get();}
}
StreamNode构造类
public class StreamNodeBuilder {public static StreamNode node(T value,StreamNodeSupplier nextNode){StreamNode streamNode = new StreamNode<>();streamNode.value = value;streamNode.nextNode = nextNode;return streamNode;}public static StreamNode startNode(StreamNodeSupplier nextNode){StreamNode streamNode = new StreamNode<>();streamNode.nextNode = nextNode;streamNode.start = true;return streamNode;}public static StreamNode endNode(T value){StreamNode streamNode = new StreamNode<>();streamNode.value = value;return streamNode;}
}
如何生成Stream
我们以一个Integer整数流的生成为例。IntStreamGen.IntStreamGen(1,10) 会返回一个流结构，其逻辑上等价于一个从1到10的整数流。但实质是一个惰性求值的stream对象，这里称其为IntStream，其StreamNodeSupplier是一个闭包，方法体是一个递归结构的求值函数，其中下界参数low = low + 1。
当IntStream第一次被求值时，流开始初始化，isStart = true。当初始化完成之后，每一次求值，都会生成一个新的流对象，其中head(low) = low + 1。当low > high时，流被终止，返回空的流对象。
public class IntStreamGen {public static StreamNode gen(int start,int end){return gen(start,end,true);}public static StreamNode gen(int start, int end, boolean isStart){//如果是开始节点，构建一个开始节点if(isStart){return StreamNodeBuilder.startNode(new StreamNodeSupplier<>(() -> gen(start,end,false)));}//判断是否结束boolean isEnd = start >= end;//生成下一节点的供应商StreamNodeSupplier nextNode = isEnd ? null : new StreamNodeSupplier<>( ()-> gen(start+1,end,false) );//返回return StreamNodeBuilder.node(start,nextNode);}}
可以看到，**生成一个流的关键在于确定如何求值下一项元素。**对于整数流来说，low = low + 1就是其下一项的求值过程。
那么对于我们非常关心的jdk集合容器，又该如何生成对应的流呢？
答案是Iterator迭代器，jdk的集合容器都实现了Iterator迭代器接口，通过迭代器我们可以轻易的取得容器的下一项元素，而不用关心容器内部实现细节。换句话说，只要实现过迭代器接口，就可以自然的转化为stream流，从而获得流计算的所有能力。
public class CollStreamGen {public static StreamNode gen(Collection coll){return gen(coll.iterator());}public static StreamNode gen(Iterator iterator){return gen(iterator,true);}private static StreamNode gen(Iterator iterator,boolean isStart){//判断是否有值boolean isEnd = !iterator.hasNext();//计算下一个值的供应商Supplier> supplier = ()->{return !iterator.hasNext() ? null : new StreamNodeSupplier<>(()-> gen(iterator,false));};if( isStart){return StreamNodeBuilder.node(null,supplier.get());}return StreamNodeBuilder.node(isEnd ? null : iterator.next(),supplier.get());}
}
实现stream的功能
在流的处理过程中添加、绑定惰性求值流程
我们以map接口举例说明。API的map接口是一个惰性求值接口，在流执行了map方法后(stream.map())，不会进行任何的求值运算。map在执行时，会生成一个新的求值过程StreamNodeSupplier，新的过程将之前流的求值过程给**“包裹"起来了，仅仅是在"流的生成"到"流的最终求值”**之间增加了一道处理工序，最终返回了一个新的stream流对象。
API.map所依赖的内部静态map方法是一个惰性求值方法，其每次调用**"只会"将当前流的head部分进行map映射操作，并且生成一个新的流。新生成流的StreamNodeSupplier和之前逻辑基本保持一致(递归)，唯一的区别是，第二个参数传入的stream在调用方法之前会被强制求值(eval)**后再传入
惰性求值接口的实现大同小异，大家需要体会一下闭包、递归、惰性求值等概念。
map
public StreamNode map(Function function){return map(function,this);
}private StreamNode map(Function function,StreamNode node){//计算当前值R newValue = node.start ? null : function.apply(node.value);//计算子节点提供者StreamNodeSupplier supplier = node.isEnd() ? null : new StreamNodeSupplier<>( ()-> map(function,node.nextNode.get()));return new StreamNode<>(newValue,supplier,node.start);
}
flatMap
public StreamNode flatMap(Function> mapper){//如果当前流是空的,直接返回if(this.isEnd())return new StreamNode(null,null,this.start);//方便 lambda 调用StreamNodeSupplier oldNodeSupplier = this.nextNode;StreamNodeSupplier newNodeSupplier = new StreamNodeSupplier<>(() -> {StreamNode streamNode = oldNodeSupplier.get();return flatMap(mapper, new StreamNode(null, null, false), streamNode);});return new StreamNode(null,newNodeSupplier,true).nextNode.get();
}private static StreamNode flatMap(Function> mapper, StreamNode headStreamNode, StreamNode streamNode){//如果头节点当前不是最后一个节点if(!headStreamNode.isEnd()){//头节点未结束,继续拼接return new StreamNode<>(headStreamNode.value, new StreamNodeSupplier<>(() -> flatMap(mapper, headStreamNode.nextNode.get(), streamNode)), false);}//如果头节点当前是最后一个节点//如果当前流结束，拼接if(streamNode.isEnd()){//计算获取新节点StreamNodeSupplier nextNodeSupplier = mapper.apply(streamNode.value).nextNode;//创建新新节点拼接2个节点return StreamNodeBuilder.node(headStreamNode.value,nextNodeSupplier);}//当前流没结束StreamNode newHeadMyStream = mapper.apply(streamNode.value).nextNode.get();//将当前节点的子节点作为头结点继续递归拼接return StreamNodeBuilder.node(headStreamNode.value,new StreamNodeSupplier<>(()->flatMap(mapper,newHeadMyStream,streamNode.nextNode.get())));}
filter
public StreamNode filter(Predicate predicate){if(this.isEnd()) return this;//闭包StreamNodeSupplier nextNodeSupplier = this.nextNode;this.nextNode = filter(predicate,nextNodeSupplier.get());return this;
}private StreamNodeSupplier filter(Predicate predicate,StreamNode node){boolean through = predicate.test(node.value);if(through){//惰性求值return new StreamNodeSupplier<>(()->{return node.isEnd() ? node : StreamNodeBuilder.node( node.value,filter( predicate,node.nextNode.get() ) );});}//递归return node.isEnd() ? null : filter(predicate,node.nextNode.get());
}
limit
public StreamNodeSupplier limit(int n,StreamNode node){if(n ==0 || node.isEnd()) return null;node.nextNode = limit(--n,node.nextNode.get());return new StreamNodeSupplier<>(()->node);
}private StreamNodeSupplier limit(int n,StreamNode node){if(n ==0 || node.isEnd()) return null;node.nextNode = limit(--n,node.nextNode.get());return new StreamNodeSupplier<>(()->node);
}
distinct
public StreamNode distinct(){//闭包StreamNodeSupplier nextNodeSupplier = this.nextNode;//惰性求值this.nextNode = new StreamNodeSupplier<>(()->distinct(new HashSet<>(),nextNodeSupplier.get()).get());return this;
}
private StreamNodeSupplier distinct(HashSet hashSet,StreamNode streamNode){final boolean contains = hashSet.contains(streamNode.value);if(contains && streamNode.isEnd()){return null;}if(contains){return distinct(hashSet,streamNode.nextNode.get());}hashSet.add(streamNode.value);//递归return new StreamNodeSupplier<>(()-> {streamNode.nextNode = distinct(hashSet,streamNode.nextNode.get());return streamNode;});
}
对流使用强制求值函数，生成最终结果
我们以forEach方法举例说明。强制求值方法forEach会不断的对当前stream进行求值并让consumer接收处理，直到当前流成为空流。
有两种可能的情况会导致递归传入的流参数成为空流(empty-stream)：
- 最初生成流的求值过程返回了空流(整数流，low > high 时，返回空流 )
- limit之类的短路操作，会提前终止流的求值返回空流(n == 0 时，返回空流)
public void forEach(Consumer function){forEach(function,this);
}private void forEach(Consumer function,StreamNode node){//如果是子节点，直接返回if(node.isEnd()) return;//获取子节点StreamNode streamNode = node.nextNode.get();//调用消费者处理function.accept(streamNode.value);//递归处理子节点if(!node.isEnd()) forEach(function,streamNode);
}
参考博客： https://blog.csdn.net/qq_33591903/article/details/110529814
上一篇：Unity 编辑器开发