1.Stream 流的介紹
1.1 java8 stream介紹
java8新增了stream流的特性，能夠讓使用者以函數式的方式、更為簡單的操縱集合等資料結構，并實作了使用者無感覺的并行計算。
1.2 從零開始實作一個stream流
相信很多人在使用過java8的streamAPI接口之後，都會對其實作原理感到好奇，但往往在看到jdk的stream源碼後卻被其複雜的抽象、封裝給弄糊塗了，而無法很好的了解其背後的原理。究其原因，是因為jdk的stream源碼是高度工程化的代碼，工程化的代碼為了效率和滿足各式各樣的需求，會将代碼實作的極其複雜，不易了解。
在這裡，我們将抛開jdk的實作思路，從零開始實作一個stream流。
我們的stream流同樣擁有惰性求值，函數式程式設計接口等特性，并相容jdk的Collection等資料結構(但不支援并行計算 orz)。
相信在親手實作一個stream流的架構之後，大家能更好的了解流計算的原理。
2.stream的優點
在探讨探究stream的實作原理和動手實作之前，我們先要體會stream流計算的獨特之處。
舉個例子： 有一個List<Person>清單,我們需要獲得年齡為70歲的前10個Person的姓名。
過程式的解決方案：
稍加思考，我們很快就寫出了一個過程式的解決方案(僞代碼)：
List<Person> personList = fromDB(); // 獲得List<Person>
int limit = 10; // 限制條件
List<String> 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<Person> personList = fromDB(); // 獲得List<Person>
List<String> 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<Person> personList = fromDB(); // 獲得List<Person>
personList.stream()
.filter(item->item.age == 70) // 過濾條件
.limit(10) // limit限制條件
.filter(item->item.name.startWith("王")) // 過濾條件
.map(item->item.name) // 獲得姓名
.forEach(System.out::println); 3.stream API接口介紹
stream API的接口是函數式的，盡管java 8也引入了lambda表達式，但java實質上依然是由接口-匿名内部類來實作函數傳參的，是以需要事先定義一系列的函數式接口。
Function: 類似于 y = F(x)
@FunctionalInterface
public interface Function<R,T> {
/**
* 函數式接口
* 類似于 y = F(x)
* */
R apply(T t);
} BiFunction: 類似于 z = F(x,y)
@FunctionalInterface
public interface BiFunction<R, T, U> {
/**
* 函數式接口
* 類似于 z = F(x,y)
* */
R apply(T t, U u);
} ForEach: 周遊處理
@FunctionalInterface
public interface ForEach <T>{
/**
* 疊代器周遊
* @param item 被疊代的每一項
* */
void apply(T item);
} Comparator: 比較器
@FunctionalInterface
public interface Comparator<T> {
/**
* 比較方法邏輯
* @param o1 參數1
* @param o2 參數2
* @return 傳回值大于0 ---> (o1 > o2)
* 傳回值等于0 ---> (o1 = o2)
* 傳回值小于0 ---> (o1 < o2)
*/
int compare(T o1, T o2);
} Predicate: 條件判斷
@FunctionalInterface
public interface Predicate <T>{
/**
* 函數式接口
* @param item 疊代的每一項
* @return true 滿足條件
* false 不滿足條件
* */
boolean satisfy(T item);
} Supplier：提供初始值
@FunctionalInterface
public interface Supplier<T> {
/**
* 提供初始值
* @return 初始化的值
* */
T get();
} EvalFunction：stream求值函數
@FunctionalInterface
public interface EvalFunction<T> {
/**
* stream流的強制求值方法
* @return 求值傳回一個新的stream
* */
MyStream<T> apply();
} stream API接口：
/**
* stream流的API接口
*/
public interface Stream<T> {
/**
* 映射 lazy 惰性求值
* @param mapper 轉換邏輯 T->R
* @return 一個新的流
* */
<R> MyStream<R> map(Function<R,T> mapper);
/**
* 扁平化 映射 lazy 惰性求值
* @param mapper 轉換邏輯 T->MyStream<R>
* @return 一個新的流(扁平化之後)
* */
<R> MyStream<R> flatMap(Function<? extends MyStream<R>, T> mapper);
/**
* 過濾 lazy 惰性求值
* @param predicate 謂詞判斷
* @return 一個新的流，其中元素是滿足predicate條件的
* */
MyStream<T> filter(Predicate<T> predicate);
/**
* 截斷 lazy 惰性求值
* @param n 截斷流，隻擷取部分
* @return 一個新的流，其中的元素不超過 n
* */
MyStream<T> limit(int n);
/**
* 去重操作 lazy 惰性求值
* @return 一個新的流，其中的元素不重複(!equals）
* */
MyStream<T> distinct();
/**
* 窺視 lazy 惰性求值
* @return 同一個流，peek不改變流的任何行為
* */
MyStream<T> peek(ForEach<T> consumer);
/**
* 周遊 eval 強制求值
* @param consumer 周遊邏輯
* */
void forEach(ForEach<T> consumer);
/**
* 濃縮 eval 強制求值
* @param initVal 濃縮時的初始值
* @param accumulator 濃縮時的 累加邏輯
* @return 濃縮之後的結果
* */
<R> R reduce(R initVal, BiFunction<R, R, T> accumulator);
/**
* 收集 eval 強制求值
* @param collector 傳入所需的函數組合子，生成高階函數
* @return 收集之後的結果
* */
<R, A> R collect(Collector<T,A,R> collector);
/**
* 最大值 eval 強制求值
* @param comparator 大小比較邏輯
* @return 流中的最大值
* */
T max(Comparator<T> comparator);
/**
* 最小值 eval 強制求值
* @param comparator 大小比較邏輯
* @return 流中的最小值
* */
T min(Comparator<T> comparator);
/**
* 計數 eval 強制求值
* @return 目前流的個數
* */
int count();
/**
* 流中是否存在滿足predicate的項
* @return true 存在 比對項
* false 不存在 比對項
* */
boolean anyMatch(Predicate<? super T> predicate);
/**
* 流中的元素是否全部滿足predicate
* @return true 全部滿足
* false 不全部滿足
* */
boolean allMatch(Predicate<? super T> predicate);
/**
* 傳回空的 stream
* @return 空stream
* */
static <T> MyStream<T> makeEmptyStream(){
// isEnd = true
return new MyStream.Builder<T>().isEnd(true).build();
}
} 4.MyStream 實作細節
簡單介紹了API接口定義之後，我們開始深入探讨流的内部實作。
流由兩個重要的部分所組成，"目前資料項(head)"和"下一資料項的求值函數(nextItemEvalProcess)"。
其中，nextItemEvalProcess是流能夠實作"惰性求值"的關鍵。
流的基本屬性：
public class MyStream<T> implements Stream<T> {
/**
* 流的頭部
* */
private T head;
/**
* 流的下一項求值函數
* */
private NextItemEvalProcess nextItemEvalProcess;
/**
* 是否是流的結尾
* */
private boolean isEnd;
public static class Builder<T>{
private MyStream<T> target;
public Builder() {
this.target = new MyStream<>();
}
public Builder<T> head(T head){
target.head = head;
return this;
}
Builder<T> isEnd(boolean isEnd){
target.isEnd = isEnd;
return this;
}
public Builder<T> nextItemEvalProcess(NextItemEvalProcess nextItemEvalProcess){
target.nextItemEvalProcess = nextItemEvalProcess;
return this;
}
public MyStream<T> build(){
return target;
}
}
/**
* 目前流強制求值
* @return 求值之後傳回一個新的流
* */
private MyStream<T> eval（){
return this.nextItemEvalProcess.eval（);
}
/**
* 目前流 為空
* */
private boolean isEmptyStream(){
return this.isEnd;
}
} /**
* 下一個元素求值過程
*/
public class NextItemEvalProcess {
/**
* 求值方法
* */
private EvalFunction evalFunction;
public NextItemEvalProcess(EvalFunction evalFunction) {
this.evalFunction = evalFunction;
}
MyStream eval（){
return evalFunction.apply();
}
} 4.1 stream流在使用過程中的三個階段
1. 生成并構造一個流 (List.stream() 等方法)
2. 在流的處理過程中添加、綁定惰性求值流程 (map、filter、limit 等方法)
3. 對流使用強制求值函數，生成最終結果 (max、collect、forEach等方法)
4.2 生成并構造一個流
流在生成時是"純淨"的，其最初的NextItemEvalProcess求值之後就是指向自己的下一個元素。
我們以一個Integer整數流的生成為例。IntegerStreamGenerator.getIntegerStream(1,10) 會傳回一個流結構，其邏輯上等價于一個從1到10的整數流。但實質是一個惰性求值的stream對象，這裡稱其為IntStream，其NextItemEvalProcess是一個閉包，方法體是一個遞歸結構的求值函數，其中下界參數low = low + 1。
當IntStream第一次被求值時，流開始初始化，isStart = false。當初始化完成之後，每一次求值，都會生成一個新的流對象，其中head(low) = low + 1。當low > high時，流被終止，傳回空的流對象。
/**
* 整數流生成器
*/
public class IntegerStreamGenerator {
/**
* 獲得一個有限的整數流 介于[low-high]之間
* @param low 下界
* @param high 上界
* */
public static MyStream<Integer> getIntegerStream(int low, int high){
return getIntegerStreamInner(low,high,true);
}
/**
* 遞歸函數。配合getIntegerStream(int low,int high)
* */
private static MyStream<Integer> getIntegerStreamInner(int low, int high, boolean isStart){
if(low > high){
// 到達邊界條件，傳回空的流
return Stream.makeEmptyStream();
}
if(isStart){
return new MyStream.Builder<Integer>()
.process(new NextItemEvalProcess(()->getIntegerStreamInner(low,high,false)))
.build();
}else{
return new MyStream.Builder<Integer>()
// 目前元素 low
.head(low)
// 下一個元素 low+1
.process(new NextItemEvalProcess(()->getIntegerStreamInner(low+1,high,false)))
.build();
}
}
} 可以看到，生成一個流的關鍵在于确定如何求值下一項元素。對于整數流來說，low = low + 1就是其下一項的求值過程。
那麼對于我們非常關心的jdk集合容器，又該如何生成對應的流呢?
答案是Iterator疊代器，jdk的集合容器都實作了Iterator疊代器接口，通過疊代器我們可以輕易的取得容器的下一項元素，而不用關心容器内部實作細節。換句話說，隻要實作過疊代器接口，就可以自然的轉化為stream流，進而獲得流計算的所有能力。
/**
* 集合流生成器
*/
public class CollectionStreamGenerator {
/**
* 将一個List轉化為stream流
* */
public static <T> MyStream<T> getListStream(List<T> list){
return getListStream(list.iterator(),true);
}
/**
* 遞歸函數
* @param iterator list 集合的疊代器
* @param isStart 是否是第一次疊代
* */
private static <T> MyStream<T> getListStream(Iterator<T> iterator, boolean isStart){
if(!iterator.hasNext()){
// 不存在疊代的下一個元素，傳回空的流
return Stream.makeEmptyStream();
}
if(isStart){
// 初始化，隻需要設定 求值過程
return new MyStream.Builder<T>()
.nextItemEvalProcess(new NextItemEvalProcess(()-> getListStream(iterator,false)))
.build();
}else{
// 非初始化，設定head和接下來的求值過程
return new MyStream.Builder<T>()
.head(iterator.next())
.nextItemEvalProcess(new NextItemEvalProcess(()-> getListStream(iterator,false)))
.build();
}
}
} 思考一個小問題，如何生成一個無窮的整數流？
4.3 在流的處理過程中添加、綁定惰性求值流程
我們以map接口舉例說明。API的map接口是一個惰性求值接口，在流執行了map方法後(stream.map())，不會進行任何的求值運算。map在執行時，會生成一個新的求值過程NextItemEvalProcess，新的過程将之前流的求值過程給"包裹"起來了，僅僅是在"流的生成"到"流的最終求值"之間增加了一道處理工序，最終傳回了一個新的stream流對象。
API.map所依賴的内部靜态map方法是一個惰性求值方法，其每次調用"隻會"将目前流的head部分進行map映射操作，并且生成一個新的流。新生成流的NextItemEvalProcess和之前邏輯基本保持一緻(遞歸)，唯一的差別是，第二個參數傳入的stream在調用方法之前會被強制求值(eval)後再傳入。
@Override
public <R> MyStream<R> map(Function<R, T> mapper) {
NextItemEvalProcess lastNextItemEvalProcess = this.nextItemEvalProcess;
this.nextItemEvalProcess = new NextItemEvalProcess(
()->{
MyStream myStream = lastNextItemEvalProcess.eval（);
return map(mapper, myStream);
}
);
// 求值鍊條 加入一個新的process map
return new MyStream.Builder<R>()
.nextItemEvalProcess(this.nextItemEvalProcess)
.build();
}
/**
* 遞歸函數 配合API.map
* */
private static <R,T> MyStream<R> map(Function<R, T> mapper, MyStream<T> myStream){
if(myStream.isEmptyStream()){
return Stream.makeEmptyStream();
}
R head = mapper.apply(myStream.head);
return new MyStream.Builder<R>()
.head(head)
.nextItemEvalProcess(new NextItemEvalProcess(()->map(mapper, myStream.eval（))))
.build();
} 惰性求值接口的實作大同小異，大家需要體會一下閉包、遞歸、惰性求值等概念，限于篇幅就不一一展開啦。
flatMap：
@Override
public <R> MyStream<R> flatMap(Function<? extends MyStream<R>,T> mapper) {
NextItemEvalProcess lastNextItemEvalProcess = this.nextItemEvalProcess;
this.nextItemEvalProcess = new NextItemEvalProcess(
()->{
MyStream myStream = lastNextItemEvalProcess.eval();
return flatMap(mapper, Stream.makeEmptyStream(), myStream);
}
);
// 求值鍊條 加入一個新的process map
return new MyStream.Builder<R>()
.nextItemEvalProcess(this.nextItemEvalProcess)
.build();
}
/**
* 遞歸函數 配合API.flatMap
* */
private static <R,T> MyStream<R> flatMap(Function<? extends MyStream<R>,T> mapper, MyStream<R> headMyStream, MyStream<T> myStream){
if(headMyStream.isEmptyStream()){
if(myStream.isEmptyStream()){
return Stream.makeEmptyStream();
}else{
T outerHead = myStream.head;
MyStream<R> newHeadMyStream = mapper.apply(outerHead);
return flatMap(mapper, newHeadMyStream.eval(), myStream.eval());
}
}else{
return new MyStream.Builder<R>()
.head(headMyStream.head)
.nextItemEvalProcess(new NextItemEvalProcess(()-> flatMap(mapper, headMyStream.eval(), myStream)))
.build();
}
} filter：
@Override
public MyStream<T> filter(Predicate<T> predicate) {
NextItemEvalProcess lastNextItemEvalProcess = this.nextItemEvalProcess;
this.nextItemEvalProcess = new NextItemEvalProcess(
()-> {
MyStream myStream = lastNextItemEvalProcess.eval（);
return filter(predicate, myStream);
}
);
// 求值鍊條 加入一個新的process filter
return this;
}
/**
* 遞歸函數 配合API.filter
* */
private static <T> MyStream<T> filter(Predicate<T> predicate, MyStream<T> myStream){
if(myStream.isEmptyStream()){
return Stream.makeEmptyStream();
}
if(predicate.satisfy(myStream.head)){
return new Builder<T>()
.head(myStream.head)
.nextItemEvalProcess(new NextItemEvalProcess(()->filter(predicate, myStream.eval（))))
.build();
}else{
return filter(predicate, myStream.eval（));
}
} limit：
@Override
public MyStream<T> limit(int n) {
NextItemEvalProcess lastNextItemEvalProcess = this.nextItemEvalProcess;
this.nextItemEvalProcess = new NextItemEvalProcess(
()-> {
MyStream myStream = lastNextItemEvalProcess.eval（);
return limit(n, myStream);
}
);
// 求值鍊條 加入一個新的process limit
return this;
}
/**
* 遞歸函數 配合API.limit
* */
private static <T> MyStream<T> limit(int num, MyStream<T> myStream){
if(num == 0 || myStream.isEmptyStream()){
return Stream.makeEmptyStream();
}
return new MyStream.Builder<T>()
.head(myStream.head)
.nextItemEvalProcess(new NextItemEvalProcess(()->limit(num-1, myStream.eval（))))
.build();
} distinct：
@Override
public MyStream<T> distinct() {
NextItemEvalProcess lastNextItemEvalProcess = this.nextItemEvalProcess;
this.nextItemEvalProcess = new NextItemEvalProcess(
()-> {
MyStream myStream = lastNextItemEvalProcess.eval（);
return distinct(new HashSet<>(), myStream);
}
);
// 求值鍊條 加入一個新的process limit
return this;
}
/**
* 遞歸函數 配合API.distinct
* */
private static <T> MyStream<T> distinct(Set<T> distinctSet,MyStream<T> myStream){
if(myStream.isEmptyStream()){
return Stream.makeEmptyStream();
}
if(!distinctSet.contains(myStream.head)){
// 加入集合
distinctSet.add(myStream.head);
return new Builder<T>()
.head(myStream.head)
.nextItemEvalProcess(new NextItemEvalProcess(()->distinct(distinctSet, myStream.eval（))))
.build();
}else{
return distinct(distinctSet, myStream.eval（));
}
} peek：
@Override
public MyStream<T> peek(ForEach<T> consumer) {
NextItemEvalProcess lastNextItemEvalProcess = this.nextItemEvalProcess;
this.nextItemEvalProcess = new NextItemEvalProcess(
()-> {
MyStream myStream = lastNextItemEvalProcess.eval（);
return peek(consumer,myStream);
}
);
// 求值鍊條 加入一個新的process peek
return this;
}
/**
* 遞歸函數 配合API.peek
* */
private static <T> MyStream<T> peek(ForEach<T> consumer,MyStream<T> myStream){
if(myStream.isEmptyStream()){
return Stream.makeEmptyStream();
}
consumer.apply(myStream.head);
return new MyStream.Builder<T>()
.head(myStream.head)
.nextItemEvalProcess(new NextItemEvalProcess(()->peek(consumer, myStream.eval（))))
.build();
} 4.4 對流使用強制求值函數，生成最終結果
我們以forEach方法舉例說明。強制求值方法forEach會不斷的對目前stream進行求值并讓consumer接收處理，直到目前流成為空流。
有兩種可能的情況會導緻遞歸傳入的流參數成為空流(empty-stream)：
1. 最初生成流的求值過程傳回了空流(整數流，low > high 時，傳回空流 )
2. limit之類的短路操作，會提前終止流的求值傳回空流(n == 0 時，傳回空流)
@Override
public void forEach(ForEach<T> consumer) {
// 終結操作 直接開始求值
forEach(consumer,this.eval（));
}
/**
* 遞歸函數 配合API.forEach
* */
private static <T> void forEach(ForEach<T> consumer, MyStream<T> myStream){
if(myStream.isEmptyStream()){
return;
}
consumer.apply(myStream.head);
forEach(consumer, myStream.eval（));
} 強制求值的接口的實作也都大同小異，限于篇幅就不一一展開啦。
reduce：
/**
* 遞歸函數 配合API.reduce
* */
private static <R,T> R reduce(R initVal, BiFunction<R,R,T> accumulator, MyStream<T> myStream){
if(myStream.isEmptyStream()){
return initVal;
}
T head = myStream.head;
R result = reduce(initVal,accumulator, myStream.eval（));
return accumulator.apply(result,head);
}
/**
* 遞歸函數 配合API.reduce
* */
private static <R,T> R reduce(R initVal, BiFunction<R,R,T> accumulator, MyStream<T> myStream){
if(myStream.isEmptyStream()){
return initVal;
}
T head = myStream.head;
R result = reduce(initVal,accumulator, myStream.eval（));
return accumulator.apply(result,head);
} max：
@Override
public T max(Comparator<T> comparator) {
// 終結操作 直接開始求值
MyStream<T> eval = this.eval（);
if(eval.isEmptyStream()){
return null;
}else{
return max(comparator,eval,eval.head);
}
}
/**
* 遞歸函數 配合API.max
* */
private static <T> T max(Comparator<T> comparator, MyStream<T> myStream, T max){
if(myStream.isEnd){
return max;
}
T head = myStream.head;
// head 和 max 進行比較
if(comparator.compare(head,max) > 0){
// head 較大 作為新的max傳入
return max(comparator, myStream.eval（),head);
}else{
// max 較大 不變
return max(comparator, myStream.eval（),max);
}
} min：
@Override
public T min(Comparator<T> comparator) {
// 終結操作 直接開始求值
MyStream<T> eval = this.eval（);
if(eval.isEmptyStream()){
return null;
}else{
return min(comparator,eval,eval.head);
}
}
/**
* 遞歸函數 配合API.min
* */
private static <T> T min(Comparator<T> comparator, MyStream<T> myStream, T min){
if(myStream.isEnd){
return min;
}
T head = myStream.head;
// head 和 min 進行比較
if(comparator.compare(head,min) < 0){
// head 較小 作為新的min傳入
return min(comparator, myStream.eval（),head);
}else{
// min 較小 不變
return min(comparator, myStream.eval（),min);
}
} count：
@Override
public int count() {
// 終結操作 直接開始求值
return count(this.eval（),0);
}
/**
* 遞歸函數 配合API.count
* */
private static <T> int count(MyStream<T> myStream, int count){
if(myStream.isEmptyStream()){
return count;
}
// count+1 進行遞歸
return count(myStream.eval（),count+1);
} anyMatch：
@Override
public boolean anyMatch(Predicate<? super T> predicate) {
// 終結操作 直接開始求值
return anyMatch(predicate,this.eval());
}
/**
* 遞歸函數 配合API.anyMatch
* */
private static <T> boolean anyMatch(Predicate<? super T> predicate,MyStream<T> myStream){
if(myStream.isEmptyStream()){
// 截止末尾，不存在任何比對項
return false;
}
// 謂詞判斷
if(predicate.satisfy(myStream.head)){
// 比對 存在比對項 傳回true
return true;
}else{
// 不比對，繼續檢查，直到存在比對項
return anyMatch(predicate,myStream.eval());
}
} allMatch：
@Override
public boolean allMatch(Predicate<? super T> predicate) {
// 終結操作 直接開始求值
return allMatch(predicate,this.eval());
}
/**
* 遞歸函數 配合API.anyMatch
* */
private static <T> boolean allMatch(Predicate<? super T> predicate,MyStream<T> myStream){
if(myStream.isEmptyStream()){
// 全部比對
return true;
}
// 謂詞判斷
if(predicate.satisfy(myStream.head)){
// 目前項比對，繼續檢查
return allMatch(predicate,myStream.eval());
}else{
// 存在不比對的項，傳回false
return false;
}
} 4.5 collect方法
collect方法是強制求值方法中，最複雜也最強大的接口，其作用是将流中的元素收集(collect)起來，并轉化成特定的資料結構。
從函數式程式設計的角度來看，collect方法是一個高階函數，其接受三個函數作為參數(supplier，accumulator，finisher)，最終生成一個更加強大的函數。在java中，三個函數參數以Collector實作對象的形式呈現。
supplier 方法：用于提供收集collect的初始值。
accumulator 方法：用于指定收集過程中，初始值和流中個體元素聚合的邏輯。
finnisher 方法：用于指定在收集完成之後的收尾轉化操作(例如：StringBuilder.toString() ---> String)。
collect接口實作：
@Override
public <R, A> R collect(Collector<T, A, R> collector) {
// 終結操作 直接開始求值
A result = collect(collector,this.eval（));
// 通過finish方法進行收尾
return collector.finisher().apply(result);
}
/**
* 遞歸函數 配合API.collect
* */
private static <R, A, T> A collect(Collector<T, A, R> collector, MyStream<T> myStream){
if(myStream.isEmptyStream()){
return collector.supplier().get();
}
T head = myStream.head;
A tail = collect(collector, myStream.eval（));
return collector.accumulator().apply(tail,head);
} collector接口：
/**
* collect接口 收集器
* 通過傳入組合子，生成高階過程
*/
public interface Collector<T, A, R> {
/**
* 收集時，提供初始化的值
* */
Supplier<A> supplier();
/**
* A = A + T
* 累加器，收集時的累加過程
* */
BiFunction<A, A, T> accumulator();
/**
* 收集完成之後的收尾操作
* */
Function<A, R> finisher();
} 了解jdk源碼的讀者可能會注意到，jdk的stream實作中collector接口多了一個combiner接口，combiner接口用于指定并行計算之後的結果集合并的邏輯，由于我們的實作不支援并行計算，是以也不需要添加combiner接口了。
同時，jdk還提供了一個Collectors工具類，很好的滿足了平時常見的需求(Collector.toList()、Collctor.groupingBy())等等。但特殊時刻還是需要使用者自己指定collect傳入的參數，精細的控制處理邏輯的，是以還是有必要了解一下collect方法内部原理的。
stream.collect()參數常用工具類：
/**
* stream.collect() 參數常用工具類
*/
public class CollectUtils {
/**
* stream 轉換為 List
* */
public static <T> Collector<T, List<T>, List<T>> toList(){
return new Collector<T, List<T>, List<T>>() {
@Override
public Supplier<List<T>> supplier() {
return ArrayList::new;
}
@Override
public BiFunction<List<T>, List<T>, T> accumulator() {
return (list, item) -> {
list.add(item);
return list;
};
}
@Override
public Function<List<T>, List<T>> finisher() {
return list -> list;
}
};
}
/**
* stream 轉換為 Set
* */
public static <T> Collector<T, Set<T>, Set<T>> toSet(){
return new Collector<T, Set<T>, Set<T>>() {
@Override
public Supplier<Set<T>> supplier() {
return HashSet::new;
}
@Override
public BiFunction<Set<T>, Set<T>, T> accumulator() {
return (set, item) -> {
set.add(item);
return set;
};
}
@Override
public Function<Set<T>, Set<T>> finisher() {
return set -> set;
}
};
}
} 4.6 舉例分析
我們選擇一個簡單而又不失一般性的例子，串聯起這些内容。通過完整的描述一個流求值的全過程，加深大家對流的了解。
public static void main(String[] args){
Integer sum = IntegerStreamGenerator.getIntegerStream(1,10)
.filter(item-> item%2 == 0) // 過濾出偶數
.map(item-> item * item) // 映射為平方
.limit(2) // 截取前兩個
.reduce(0,(i1,i2)-> i1+i2); // 最終結果累加求和(初始值為0)
System.out.println(sum); // 20
} 由于我們的stream實作采用的是鍊式程式設計的方式，不太好了解，将其展開為邏輯等價的形式。
public static void main(String[] args){
// 生成整數流 1-10
Stream<Integer> intStream = IntegerStreamGenerator.getIntegerStream(1,10);
// intStream基礎上過濾出偶數
Stream<Integer> filterStream = intStream.filter(item-> item%2 == 0);
// filterStream基礎上映射為平方
Stream<Integer> mapStream = filterStream.map(item-> item * item);
// mapStream基礎上截取前兩個
Stream<Integer> limitStream = mapStream.limit(2);
// 最終結果累加求和(初始值為0)
Integer sum = limitStream.reduce(0,(i1,i2)-> i1+i2);
System.out.println(sum); // 20
} reduce強制求值操作之前的執行過程圖：
reduce強制求值過程中的執行過程圖 ：
可以看到，stream的求值過程并不會一口氣将初始的流全部求值，而是按需的、一個一個的進行求值。
stream的一次求值過程至多隻會周遊流中元素一次；如果存在短路操作(limit、anyMatch等)，實際疊代的次數會更少。
是以不必擔心多層的map、filter處理邏輯的嵌套會讓流進行多次疊代，導緻效率急劇下降。
5.stream 總結
5.1 目前版本缺陷
1. 遞歸調用效率較低
為了代碼的簡潔性和更加的函數式，目前實作中很多地方都用遞歸代替了循環疊代。
雖然邏輯上遞歸和疊代是等價的，但在目前的計算機硬體上，每一層的遞歸調用都會使得函數調用棧增大，而即使是明顯的尾遞歸調用，java目前也沒有能力進行優化。當流需要處理的資料量很大時，将會出現棧溢出，棧空間不足之類的系統錯誤。
将遞歸優化為疊代能夠顯著提高目前版本流的執行效率。
2. API接口較少
限于篇幅，我們隻提供了一些較為常用的API接口。在jdk中，Collector工具類提供了很多友善易用的接口；對于同一API接口也提供了多種重載函數給使用者使用。
以目前已有的功能為基礎，提供一些更加友善的接口并不困難。
3. 不支援并行計算
由于流在求值計算時生成的是對象的副本，是無副作用的，很适合通過資料分片執行并行計算。限于個人水準，在設計之初并沒有考慮将并行計算這一特性加入進來。
5.2 函數式程式設計
仔細分析整個流的執行過程，與其說流是一個對象，不如說流是一個高階函數(higher-order function)。每當map、filter綁定了一個流，新生成的流其實是一個更加複雜的函數；每一層封裝，都會使新生成的流這一高階函數比起原基礎變得更加強大和複雜。map、filter就像一個個的基礎算子，在接收對應的過程後(filter(過濾出偶數)、map(平方映射))，可以不斷的疊加，完成許許多多非常複雜的操作。
這也是函數式程式設計的中心思想之一：将計算過程轉化為一系列嵌套函數的調用。
5.3 總結
最初是在學習《計算機程式的構造和解釋》(SICP)中stream流計算時突發奇想的，想着能不能用java來實作一個和書上類似的流計算架構，能和jdk的stream流功能大緻相同，最終，通過反複地思考和嘗試才将心中所想以java代碼的形式呈現出來。