T是流中要收集的項(xiàng)目的泛型。

A是累加器的類型，累加器是在收集過程中用于累積部分結(jié)果的對(duì)象。

R是收集操作得到的對(duì)象（通常但并不一定是集合）的類型。

例如，你可以實(shí)現(xiàn)一個(gè)ToListCollector類，將Stream中的所有元素收集List里，它的簽名如下：

public class ToListCollector implements Collector, List>

在調(diào)用時(shí)它會(huì)創(chuàng)建一個(gè)空的累加器實(shí)例，供數(shù)據(jù)收集過程使用

public Supplier> supplier() { 
 return () -> new ArrayList(); 
}

或者使用構(gòu)造函數(shù)引用;

public Supplier> supplier() { 
 return ArrayList::new; 
}

accumulator方法會(huì)返回執(zhí)行歸約操作的函數(shù)。當(dāng)遍歷到流中第n個(gè)元素時(shí)，這個(gè)函數(shù)執(zhí)行時(shí)會(huì)有兩個(gè)參數(shù)：保存歸約結(jié)果的累加器（已收集了流中的前 n?1 個(gè)項(xiàng)目），還有第n個(gè)元素本身。該函數(shù)將返回void，因?yàn)槔奂悠魇窃桓拢春瘮?shù)的執(zhí)行改變了它的內(nèi)部狀態(tài)以體現(xiàn)遍歷的元素的效果。對(duì)于ToListCollector，這個(gè)函數(shù)僅僅會(huì)把當(dāng)前項(xiàng)目添加至已經(jīng)遍歷過的項(xiàng)目的列表：

public BiConsumer, T> accumulator() { 
 return (list, item) -> list.add(item); 
} 

你也可以使用方法引用，這會(huì)更為簡(jiǎn)潔：

public BiConsumer, T> accumulator() { 
 return List::add; 
}

在遍歷完流后，finisher方法必須返回在累積過程的最后要調(diào)用的一個(gè)函數(shù)，以便將累加器對(duì)象轉(zhuǎn)換為整個(gè)集合操作的最終結(jié)果。

public Function, List> finisher() { 
 return Function.identity(); //累加器對(duì)象恰好符合預(yù)期的最終結(jié)果，
//因此無需進(jìn)行轉(zhuǎn)換。所以finisher方法只需返回identity函數(shù)
}

combiner方法會(huì)返回一個(gè)供歸約操作使用的函數(shù)，它定義了對(duì)流的各個(gè)子部分進(jìn)行并行處理時(shí)，各個(gè)子部分歸約所得的累加器要如何合并。

public BinaryOperator> combiner() { 
 return (list1, list2) -> { 
 list1.addAll(list2); 
 return list1; } 
}

有了這第四個(gè)方法，就可以對(duì)流進(jìn)行并行歸約了,會(huì)用到Java 7中引入的Fork/Join框架和Spliterator抽象

Fork/Join是什么？
Fork/Join框架是Java7提供的并行執(zhí)行任務(wù)框架，思想是將大任務(wù)分解成小任務(wù)，然后小任務(wù)又可以繼續(xù)分解，然后每個(gè)小任務(wù)分別計(jì)算出結(jié)果再合并起來，最后將匯總的結(jié)果作為大任務(wù)結(jié)果。其思想和MapReduce的思想非常類似。對(duì)于任務(wù)的分割，要求各個(gè)子任務(wù)之間相互獨(dú)立，能夠并行獨(dú)立地執(zhí)行任務(wù)，互相之間不影響。
Fork/Join的運(yùn)行流程圖如下：

我們可以通過Fork/Join單詞字面上的意思去理解這個(gè)框架。Fork是叉子分叉的意思，即將大任務(wù)分解成并行的小任務(wù)，Join是連接結(jié)合的意思，即將所有并行的小任務(wù)的執(zhí)行結(jié)果匯總起來。

工作竊取算法
ForkJoin采用了工作竊取（work-stealing）算法，若一個(gè)工作線程的任務(wù)隊(duì)列為空沒有任務(wù)執(zhí)行時(shí)，便從其他工作線程中獲取任務(wù)主動(dòng)執(zhí)行。為了實(shí)現(xiàn)工作竊取，在工作線程中維護(hù)了雙端隊(duì)列，竊取任務(wù)線程從隊(duì)尾獲取任務(wù)，被竊取任務(wù)線程從隊(duì)頭獲取任務(wù)。這種機(jī)制充分利用線程進(jìn)行并行計(jì)算，減少了線程競(jìng)爭(zhēng)。但是當(dāng)隊(duì)列中只存在一個(gè)任務(wù)了時(shí)，兩個(gè)線程去取反而會(huì)造成資源浪費(fèi)。
工作竊取的運(yùn)行流程圖如下：

Fork/Join核心類
1.ForkJoinPool
ForkJoinPool是ForkJoin框架中的任務(wù)調(diào)度器，和ThreadPoolExecutor一樣實(shí)現(xiàn)了自己的線程池，提供了三種調(diào)度子任務(wù)的方法：
execute：異步執(zhí)行指定任務(wù)，無返回結(jié)果；
invoke、invokeAll：同步執(zhí)行指定任務(wù)，等待完成才返回結(jié)果；
submit：異步執(zhí)行指定任務(wù)，并立即返回一個(gè)Future對(duì)象；
2.ForkJoinTask
Fork/Join框架中的實(shí)際的執(zhí)行任務(wù)類，有以下兩種實(shí)現(xiàn)，一般繼承這兩種實(shí)現(xiàn)類即可。
RecursiveAction：用于無結(jié)果返回的子任務(wù)；
RecursiveTask：用于有結(jié)果返回的子任務(wù)；
Fork/Join框架實(shí)戰(zhàn)
下面實(shí)現(xiàn)一個(gè)Fork/Join小例子，從1+2+...10億，每個(gè)任務(wù)只能處理1000個(gè)數(shù)相加，超過1000個(gè)的自動(dòng)分解成小任務(wù)并行處理；并展示了通過不使用Fork/Join和使用時(shí)的時(shí)間損耗對(duì)比。
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

public class ForkJoinTask extends RecursiveTask {
   private static final long MAX = 1000000000L;
   private static final long THRESHOLD = 1000L;
   private long start;
   private long end;

   public ForkJoinTask(long start, long end) {
       this.start = start;
       this.end = end;
   }

   public static void main(String[] args) {
       test();
       System.out.println("--------------------");
       testForkJoin();
   }

   private static void test() {
       System.out.println("test");
       long start = System.currentTimeMillis();
       Long sum = 0L;
       for (long i = 0L; i <= MAX; i++) {
           sum += i;
       }
       System.out.println(sum);
       System.out.println(System.currentTimeMillis() - start + "ms");
   }

   private static void testForkJoin() {
       System.out.println("testForkJoin");
       long start = System.currentTimeMillis();
       ForkJoinPool forkJoinPool = new ForkJoinPool();
       Long sum = forkJoinPool.invoke(new ForkJoinTask(1, MAX));
       System.out.println(sum);
       System.out.println(System.currentTimeMillis() - start + "ms");
   }

   @Override
   protected Long compute() {
       long sum = 0;
       if (end - start <= THRESHOLD) {
           for (long i = start; i <= end; i++) {
               sum += i;
           }
           return sum;
       } else {
           long mid = (start + end) / 2;

           ForkJoinTask task1 = new ForkJoinTask(start, mid);
           task1.fork();

           ForkJoinTask task2 = new ForkJoinTask(mid + 1, end);
           task2.fork();

           return task1.join() + task2.join();
       }
   }

}
這里需要計(jì)算結(jié)果，所以任務(wù)繼承的是RecursiveTask類。ForkJoinTask需要實(shí)現(xiàn)compute方法，在這個(gè)方法里首先需要判斷任務(wù)是否小于等于閾值1000，如果是就直接執(zhí)行任務(wù)。否則分割成兩個(gè)子任務(wù)，每個(gè)子任務(wù)在調(diào)用fork方法時(shí)，又會(huì)進(jìn)入compute方法，看看當(dāng)前子任務(wù)是否需要繼續(xù)分割成孫任務(wù)，如果不需要繼續(xù)分割，則執(zhí)行當(dāng)前子任務(wù)并返回結(jié)果。使用join方法會(huì)阻塞并等待子任務(wù)執(zhí)行完并得到其結(jié)果。
程序輸出：
test
500000000500000000
4992ms
--------------------
testForkJoin
500000000500000000
508ms
需要特別注意的是：
ForkJoinPool 使用submit 或 invoke 提交的區(qū)別：invoke是同步執(zhí)行，調(diào)用之后需要等待任務(wù)完成，才能執(zhí)行后面的代碼；submit是異步執(zhí)行，只有在Future調(diào)用get的時(shí)候會(huì)阻塞。
這里繼承的是RecursiveTask，還可以繼承RecursiveAction。前者適用于有返回值的場(chǎng)景，而后者適合于沒有返回值的場(chǎng)景
這一點(diǎn)是最容易忽略的地方，其實(shí)這里執(zhí)行子任務(wù)調(diào)用fork方法并不是最佳的選擇，最佳的選擇是invokeAll方法。
leftTask.fork();  
rightTask.fork();
替換為
invokeAll(leftTask, rightTask);
具體說一下原理：對(duì)于Fork/Join模式，假如Pool里面線程數(shù)量是固定的，那么調(diào)用子任務(wù)的fork方法相當(dāng)于A先分工給B，然后A當(dāng)監(jiān)工不干活，B去完成A交代的任務(wù)。所以上面的模式相當(dāng)于浪費(fèi)了一個(gè)線程。那么如果使用invokeAll相當(dāng)于A分工給B后，A和B都去完成工作。這樣可以更好的利用線程池，縮短執(zhí)行的時(shí)間。

返回一個(gè)不可變的Characteristics集合，它定義了收集器的行為——尤其是關(guān)于流是否可以并行歸約，以及可以使用哪些優(yōu)化的提示。
Characteristics是一個(gè)包含三個(gè)項(xiàng)目的枚舉。

UNORDERED——?dú)w約結(jié)果不受流中項(xiàng)目的遍歷和累積順序的影響。

CONCURRENT——accumulator函數(shù)可以從多個(gè)線程同時(shí)調(diào)用，且該收集器可以并行歸約流。如果收集器沒有標(biāo)為UNORDERED，那它僅在用于無序數(shù)據(jù)源時(shí)才可以并行歸約。

IDENTITY_FINISH——這表明完成器方法返回的函數(shù)是一個(gè)恒等函數(shù)，可以跳過。這種情況下，累加器對(duì)象將會(huì)直接用作歸約過程的最終結(jié)果。這也意味著，將累加器A不加檢查地轉(zhuǎn)換為結(jié)果R是安全的。

@Override 
 public Set characteristics() { 
 return Collections.unmodifiableSet(EnumSet.of( 
 IDENTITY_FINISH, CONCURRENT));
 }

Stream有一個(gè)重載的collect方法可以接受另外三個(gè)函數(shù)——supplier、accumulator和combiner，其語義和Collector接口的相應(yīng)方法返回的函數(shù)完全相同。

List dishes = menuStream.collect( 
 ArrayList::new,
 List::add,
 List::addAll);//它永遠(yuǎn)都是一個(gè)IDENTITY_FINISH和CONCURRENT但并非UNORDERED的收集器。