摘要：分區函數返回一個布爾值，這意味著得到的分組的鍵類型是，于是它最多可以分為兩組是一組，是一組。當遍歷到流中第個元素時，這個函數執行時會有兩個參數保存歸約結果的累加器已收集了流中的前個項目，還有第個元素本身。
一、收集器簡介
把列表中的交易按貨幣分組：
Map> transactionsByCurrencies =
transactions.stream().collect(groupingBy(Transaction::getCurrency));
從Collectors
類提供的工廠方法（例如groupingBy）創建的收集器。它們主要提供了三大功能：
將流元素歸約和匯總為一個值
元素分組
元素分區
二、歸約和匯總
數一數菜單里有多少種菜：
long howManyDishes = menu.stream().collect(Collectors.counting());
這還可以寫得更為直接：
long howManyDishes = menu.stream().count();
1.查找流中的最大值和最小值
可以使用兩個收集器，Collectors.maxBy和Collectors.minBy，來計算流中的最大或最小值。這兩個收集器接收一個Comparator參數來比較流中的元素.
找出菜單中熱量最高的菜:
Comparator dishCaloriesComparator =
Comparator.comparingInt(Dish::getCalories);
Optional mostCalorieDish =
menu.stream()
.collect(maxBy(dishCaloriesComparator));
2.匯總
Collectors.summingInt
它可接受一個把對象映射為求和所需int的函數，并返回一個收集器；該收集器在傳遞給普通的collect方法后即執行我們需要的匯總操作。
eg:
int totalCalories = menu.stream().collect(summingInt(Dish::getCalories));
另外，Collectors.summingLong和Collectors.summingDouble方法的作用完全一樣，可以用于求和字段為long或double的情況。還有Collectors.averagingInt，連同對應的averagingLong和averagingDouble可以計算數值的平均數。
summarizing操作
通過一次summarizing操作你可以就數出菜單中元素的個數，并得到菜肴熱量總和、平均值、最大值和最小值：
IntSummaryStatistics menuStatistics =
menu.stream().collect(summarizingInt(Dish::getCalories));
這個收集器會把所有這些信息收集到一個叫作IntSummaryStatistics的類里，它提供了方便的取值（getter）方法來訪問結果。打印menuStatisticobject會得到以下輸出：
IntSummaryStatistics{count=9, sum=4300, min=120,
average=477.777778, max=800}
同樣，相應的summarizingLong和summarizingDouble工廠方法有相關的LongSummaryStatistics和DoubleSummaryStatistics類型。
3.連接字符串
joining工廠方法返回的收集器會把對流中每一個對象應用toString方法得到的所有字符串連接成一個字符串。
String shortMenu = menu.stream().map(Dish::getName).collect(joining());
joining工廠方法有一個重載版本可以接受元素之間的分界符
String shortMenu = menu.stream().map(Dish::getName).collect(joining(", "));
4.廣義的歸約匯總
可以用reducing方法創建的收集器來計算你菜單的總熱量，如下所示：
int totalCalories = menu.stream().collect(reducing(
0, Dish::getCalories, (i, j) -> i + j));
第一個參數是歸約操作的起始值。
第二個參數將菜肴轉換成一個表示其所含熱量的int。
第三個參數是一個BinaryOperator，將兩個項目累積成一個同類型的值。這里它就是對兩個int求和。
單參數形式的reducing來找到熱量最高的菜，如下所示：
Optional mostCalorieDish =
menu.stream().collect(reducing(
(d1, d2) -> d1.getCalories() > d2.getCalories() ? d1 : d2));
相比stream的reduce方法collect方法特別適合表達可變容器上的歸約，更關鍵的是它適合并行操作
計算菜單里所有菜肴的卡路里總和，以不同的方法執行同樣的操作:
第一種：
int totalCalories = menu.stream().collect(reducing(0,
Dish::getCalories,
Integer::sum));
第二種：
int totalCalories =
menu.stream().map(Dish::getCalories).reduce(Integer::sum).get();//reduce返回的是Optional
第三種:
int totalCalories = menu.stream().mapToInt(Dish::getCalories).sum();
最后一種最佳。
三、分組
假設你要把菜單中的菜按照類型進行分類，有肉的放一組，有魚的放一組，其他的都放另一組。用Collectors.groupingBy工廠方法返回的收集器就可以輕松地完成這項任務，如下所示：
Map> dishesByType =
menu.stream().collect(groupingBy(Dish::getType));
其結果是下面的Map：
{FISH=[prawns, salmon], OTHER=[french fries, rice, season fruit, pizza],
MEAT=[pork, beef, chicken]}
給groupingBy方法傳遞了一個Function（以方法引用的形式），它提取了流中每一道Dish的Dish.Type。我們把這個Function叫作分類函數
如果Dish中沒有定義類型獲取方法，可以使用lambda表達式：
public enum CaloricLevel { DIET, NORMAL, FAT }
Map> dishesByCaloricLevel = menu.stream().collect(
groupingBy(dish -> {
if (dish.getCalories() <= 400) return CaloricLevel.DIET;
else if (dish.getCalories() <= 700) return
CaloricLevel.NORMAL;
else return CaloricLevel.FAT;
} ));
1.多級分組
使用一個由雙參數版本的Collectors.groupingBy工廠方法創建的收集器，它除了普通的分類函數之外，還可以接受collector類型的第二個參數:
Map>> dishesByTypeCaloricLevel =
menu.stream().collect(
groupingBy(Dish::getType,
groupingBy(dish -> {
if (dish.getCalories() <= 400) return CaloricLevel.DIET;
else if (dish.getCalories() <= 700) return CaloricLevel.NORMAL;
else return CaloricLevel.FAT;
} )
)
);
這種多級分組操作可以擴展至任意層級，n級分組就會得到一個代表n級樹形結構的n級Map
2.按子組收集數據
傳遞給第一個groupingBy的第二個收集器可以是任何類型，而不一定是另一groupingBy
Map typesCount = menu.stream().collect(
groupingBy(Dish::getType, counting()));
其結果是下面的Map：
{MEAT=3, FISH=2, OTHER=4}
普通的單參數groupingBy(f)（其中f是分類函數）實際上是groupingBy(f, toList())的簡便寫法。
把收集器的結果轉換為另一種類型
查找每個子組中熱量最高的Dish
Map mostCaloricByType =
menu.stream()
.collect(groupingBy(Dish::getType,
collectingAndThen(
maxBy(comparingInt(Dish::getCalories)), //maxBy工廠方法生成的收集器的類型是Optional
Optional::get)));
包裝的Optional沒什么用，把收集器返回的結果轉換為另一種類型，你可以使用Collectors.collectingAndThen工廠方法；返回的收集器groupingBy收集器只有在應用分組條件后，第一次在流中找到某個鍵對應的元素時才會把鍵加入分組Map中,所以Optional::get這個操作放在這里是安全的，因為reducing收集器永遠都不會返回Optional.empty()
與groupingBy聯合使用的其他收集器的例子
Map totalCaloriesByType =
menu.stream().collect(groupingBy(Dish::getType,
summingInt(Dish::getCalories)));
對于每種類型的Dish，菜單中都有哪些CaloricLevel。我們可以把groupingBy和mapping收集器結合起來，如下所示：
Map> caloricLevelsByType =
menu.stream().collect(
groupingBy(Dish::getType, mapping(
dish -> {
if (dish.getCalories() <= 400) return CaloricLevel.DIET;
else if (dish.getCalories() <= 700) return CaloricLevel.NORMAL;
else return CaloricLevel.FAT;
},
toSet() )));//生成的CaloricLevel流傳遞給一個toSet收集器，
//它和toList類似，不過是把流中的元素累積到一個Set而不是List中，以便僅保留各不相同的值。
但通過使用toCollection，你就可以有更多的控制。例如，你可以給它傳遞一個構造函數引用來要求HashSet:
Map> caloricLevelsByType =
menu.stream().collect(
groupingBy(Dish::getType, mapping(
dish -> { if (dish.getCalories() <= 400) return CaloricLevel.DIET;
else if (dish.getCalories() <= 700) return CaloricLevel.NORMAL;
else return CaloricLevel.FAT; },
toCollection(HashSet::new) )));
四、分區
1.分區的優勢
分區是分組的特殊情況：由一個謂詞（返回一個布爾值的函數）作為分類函數，它稱分區函數。分區函數返回一個布爾值，這意味著得到的分組Map的鍵類型是Boolean，于是它最多可以分為兩組——true是一組，false是一組。例如，如果你是素食者或是請了一位素食的朋友來共進晚餐，可能會想要把菜單按照素食和非素食分開：
Map> partitionedMenu =
menu.stream().collect(partitioningBy(Dish::isVegetarian));
計算素食和非素食的數量：
menu.stream().collect(partitioningBy(Dish::isVegetarian,
counting()));
2.將數字按質數和非質數分區
public boolean isPrime(int candidate) {
int candidateRoot = (int) Math.sqrt((double) candidate);
return IntStream.rangeClosed(2, candidateRoot)
.noneMatch(i -> candidate % i == 0);
}
public Map> partitionPrimes(int n) {
return IntStream.rangeClosed(2, n).boxed()
.collect(
partitioningBy(candidate -> isPrime(candidate)));
}
Collectors類的靜態工廠方法:
五、收集器接口
public interface Collector {
Supplier supplier();
BiConsumer accumulator();
Function finisher();
BinaryOperator combiner();
Set characteristics();
}
T是流中要收集的項目的泛型。
A是累加器的類型，累加器是在收集過程中用于累積部分結果的對象。
R是收集操作得到的對象（通常但并不一定是集合）的類型。
例如，你可以實現一個ToListCollector類，將Stream中的所有元素收集List里，它的簽名如下：
public class ToListCollector implements Collector, List>
1.理解 Collector 接口聲明的方法
(1)建立新的結果容器：supplier方法
在調用時它會創建一個空的累加器實例，供數據收集過程使用
public Supplier> supplier() {
return () -> new ArrayList();
}
或者使用構造函數引用;
public Supplier> supplier() {
return ArrayList::new;
}
(2)將元素添加到結果容器：accumulator方法
accumulator方法會返回執行歸約操作的函數。當遍歷到流中第n個元素時，這個函數執行時會有兩個參數：保存歸約結果的累加器（已收集了流中的前 n?1 個項目），還有第n個元素本身。該函數將返回void，因為累加器是原位更新，即函數的執行改變了它的內部狀態以體現遍歷的元素的效果。對于ToListCollector，這個函數僅僅會把當前項目添加至已經遍歷過的項目的列表：
public BiConsumer, T> accumulator() {
return (list, item) -> list.add(item);
}
你也可以使用方法引用，這會更為簡潔：
public BiConsumer, T> accumulator() {
return List::add;
}
(3)對結果容器應用最終轉換：finisher方法
在遍歷完流后，finisher方法必須返回在累積過程的最后要調用的一個函數，以便將累加器對象轉換為整個集合操作的最終結果。
public Function, List> finisher() {
return Function.identity(); //累加器對象恰好符合預期的最終結果，
//因此無需進行轉換。所以finisher方法只需返回identity函數
}
(4) 合并兩個結果容器：combiner方法
combiner方法會返回一個供歸約操作使用的函數，它定義了對流的各個子部分進行并行處理時，各個子部分歸約所得的累加器要如何合并。
public BinaryOperator> combiner() {
return (list1, list2) -> {
list1.addAll(list2);
return list1; }
}
有了這第四個方法，就可以對流進行并行歸約了,會用到Java 7中引入的Fork/Join框架和Spliterator抽象
Fork/Join是什么？
Fork/Join框架是Java7提供的并行執行任務框架，思想是將大任務分解成小任務，然后小任務又可以繼續分解，然后每個小任務分別計算出結果再合并起來，最后將匯總的結果作為大任務結果。其思想和MapReduce的思想非常類似。對于任務的分割，要求各個子任務之間相互獨立，能夠并行獨立地執行任務，互相之間不影響。
Fork/Join的運行流程圖如下：
我們可以通過Fork/Join單詞字面上的意思去理解這個框架。Fork是叉子分叉的意思，即將大任務分解成并行的小任務，Join是連接結合的意思，即將所有并行的小任務的執行結果匯總起來。
工作竊取算法
ForkJoin采用了工作竊取（work-stealing）算法，若一個工作線程的任務隊列為空沒有任務執行時，便從其他工作線程中獲取任務主動執行。為了實現工作竊取，在工作線程中維護了雙端隊列，竊取任務線程從隊尾獲取任務，被竊取任務線程從隊頭獲取任務。這種機制充分利用線程進行并行計算，減少了線程競爭。但是當隊列中只存在一個任務了時，兩個線程去取反而會造成資源浪費。
工作竊取的運行流程圖如下：
Fork/Join核心類
1.ForkJoinPool
ForkJoinPool是ForkJoin框架中的任務調度器，和ThreadPoolExecutor一樣實現了自己的線程池，提供了三種調度子任務的方法：
execute：異步執行指定任務，無返回結果；
invoke、invokeAll：同步執行指定任務，等待完成才返回結果；
submit：異步執行指定任務，并立即返回一個Future對象；
2.ForkJoinTask
Fork/Join框架中的實際的執行任務類，有以下兩種實現，一般繼承這兩種實現類即可。
RecursiveAction：用于無結果返回的子任務；
RecursiveTask：用于有結果返回的子任務；
Fork/Join框架實戰
下面實現一個Fork/Join小例子，從1+2+...10億，每個任務只能處理1000個數相加，超過1000個的自動分解成小任務并行處理；并展示了通過不使用Fork/Join和使用時的時間損耗對比。
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();
}
}
}
這里需要計算結果，所以任務繼承的是RecursiveTask類。ForkJoinTask需要實現compute方法，在這個方法里首先需要判斷任務是否小于等于閾值1000，如果是就直接執行任務。否則分割成兩個子任務，每個子任務在調用fork方法時，又會進入compute方法，看看當前子任務是否需要繼續分割成孫任務，如果不需要繼續分割，則執行當前子任務并返回結果。使用join方法會阻塞并等待子任務執行完并得到其結果。
程序輸出：
test
500000000500000000
4992ms
--------------------
testForkJoin
500000000500000000
508ms
需要特別注意的是：
ForkJoinPool 使用submit 或 invoke 提交的區別：invoke是同步執行，調用之后需要等待任務完成，才能執行后面的代碼；submit是異步執行，只有在Future調用get的時候會阻塞。
這里繼承的是RecursiveTask，還可以繼承RecursiveAction。前者適用于有返回值的場景，而后者適合于沒有返回值的場景
這一點是最容易忽略的地方，其實這里執行子任務調用fork方法并不是最佳的選擇，最佳的選擇是invokeAll方法。
leftTask.fork();
rightTask.fork();
替換為
invokeAll(leftTask, rightTask);
具體說一下原理：對于Fork/Join模式，假如Pool里面線程數量是固定的，那么調用子任務的fork方法相當于A先分工給B，然后A當監工不干活，B去完成A交代的任務。所以上面的模式相當于浪費了一個線程。那么如果使用invokeAll相當于A分工給B后，A和B都去完成工作。這樣可以更好的利用線程池，縮短執行的時間。
(5) characteristics方法
返回一個不可變的Characteristics集合，它定義了收集器的行為——尤其是關于流是否可以并行歸約，以及可以使用哪些優化的提示。
Characteristics是一個包含三個項目的枚舉。
UNORDERED——歸約結果不受流中項目的遍歷和累積順序的影響。
CONCURRENT——accumulator函數可以從多個線程同時調用，且該收集器可以并行歸約流。如果收集器沒有標為UNORDERED，那它僅在用于無序數據源時才可以并行歸約。
IDENTITY_FINISH——這表明完成器方法返回的函數是一個恒等函數，可以跳過。這種情況下，累加器對象將會直接用作歸約過程的最終結果。這也意味著，將累加器A不加檢查地轉換為結果R是安全的。
@Override
public Set characteristics() {
return Collections.unmodifiableSet(EnumSet.of(
IDENTITY_FINISH, CONCURRENT));
}
2.進行自定義收集而不去實現Collector
Stream有一個重載的collect方法可以接受另外三個函數——supplier、accumulator和combiner，其語義和Collector接口的相應方法返回的函數完全相同。
List dishes = menuStream.collect(
ArrayList::new,
List::add,
List::addAll);//它永遠都是一個IDENTITY_FINISH和CONCURRENT但并非UNORDERED的收集器。
