本列表適用以下定義。

T 是流中要收集的項目的泛型。

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

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

例如，你可以實現一個 ToListCollector 類，將 Stream 中的所有元素收集到一個List 里，它的簽名如下：

public class ToListCollector implements Collector, List>

我們很快就會澄清，這里用于累積的對象也將是收集過程的最終結果。

現在我們可以一個個來分析 Collector 接口聲明的五個方法了。通過分析，你會注意到，前四個方法都會返回一個會被 collect 方法調用的函數，而第五個方法 characteristics 則提供了一系列特征，也就是一個提示列表，告訴 collect 方法在執行歸約操作的時候可以應用哪些優化（比如并行化）。

1. 建立新的結果容器： supplier 方法

supplier 方法必須返回一個結果為空的 Supplier ，也就是一個無參數函數，在調用時它會創建一個空的累加器實例，供數據收集過程使用。很明顯，對于將累加器本身作為結果返回的收集器，比如我們的 ToListCollector ，在對空流執行操作的時候，這個空的累加器也代表了收集過程的結果。在我們的 ToListCollector 中， supplier 返回一個空的 List ，如下所示：

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

請注意你也可以只傳遞一個構造函數引用：

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

2. 將元素添加到結果容器： accumulator 方法

accumulator 方法會返回執行歸約操作的函數。當遍歷到流中第n個元素時，這個函數執行時會有兩個參數：保存歸約結果的累加器（已收集了流中的前 n-1 個項目），還有第n個元素本身。該函數將返回void ，因為累加器是原位更新，即函數的執行改變了它的內部狀態以體現遍歷的元素的效果。對于ToListCollector ，這個函數僅僅會把當前項目添加至已經遍歷過的項目的列表：

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

你也可以使用方法引用，這會更為簡潔：

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

3. 對結果容器應用最終轉換： finisher 方法

在遍歷完流后， finisher 方法必須返回在累積過程的最后要調用的一個函數，以便將累加器對象轉換為整個集合操作的最終結果。通常，就像 ToListCollector 的情況一樣，累加器對象恰好符合預期的最終結果，因此無需進行轉換。所以 finisher 方法只需返回 identity 函數：

@Override
public Function, List> finisher() {
    return Function.identity();
}

這三個方法已經足以對流進行循序規約。實踐中的實現細節可能還要復雜一點，一方面是應為流的延遲性質，可能在collect操作之前還需完成其他中間操作的流水線，另一方面則是理論上可能要進行并行規約。

4. 合并兩個結果容器： combiner 方法

四個方法中的最后一個————combiner方法會返回一個供歸約操作的使用函數，它定義了對流的各個子部分進行并行處理時，各個子部分歸約所得的累加器要如何合并。對于toList而言，這個方法的實現非常簡單，只要把從流的第二個部分收集到的項目列表加到遍歷第一部分時得到的列表后面就行了：

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

有了這第四個方法，就可以對流進行并行歸約了。它會用到Java7中引入的分支/合并框架和Spliterator抽象。

5. characteristics 方法

最后一個方法—— characteristics 會返回一個不可變的 Characteristics 集合，它定義了收集器的行為——尤其是關于流是否可以并行歸約，以及可以使用哪些優化的提示。Characteristics 是一個包含三個項目的枚舉。

UNORDERED ——歸約結果不受流中項目的遍歷和累積順序的影響。

CONCURRENT —— accumulator 函數可以從多個線程同時調用，且該收集器可以并行歸約流。如果收集器沒有標為 UNORDERED ，那它僅在用于無序數據源時才可以并行歸約。

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

我們迄今開發的 ToListCollector 是 IDENTITY_FINISH 的，因為用來累積流中元素的List 已經是我們要的最終結果，用不著進一步轉換了，但它并不是 UNORDERED ，因為用在有序流上的時候，我們還是希望順序能夠保留在得到的 List 中。最后，它是 CONCURRENT 的，但我們剛才說過了，僅僅在背后的數據源無序時才會并行處理。

前一小節中談到的五個方法足夠我們開發自己的 ToListCollector 了。你可以把它們都融合起來，如下面的代碼清單所示。

public class ToListCollector implements Collector, List> {
    @Override
    public Supplier> supplier() {
        return ArrayList::new;
    }

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

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

    @Override
    public Function, List> finisher() {
        return Function.identity();
    }

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

請注意，這個是實現與Collections.toList()方法并不完全相同，但區別僅僅是一些小的優化。這些優化的一個主要方面是Java API所提供的收集器在需要返回空列表時使用了 Collections.emptyList() 這個單例（singleton）。這意味著它可安全地替代原生Java，來收集菜單流中的所有 Dish 的列表：

List dishes = menuStream.collect(new ToListCollector<>());

這個實現和標準的

List dishes = menuStream.collect(toList());

構造之間的其他差異在于 toList 是一個工廠，而 ToListCollector 必須用 new 來實例化。

進行自定義收集而不去實現 Collector

對于 IDENTITY_FINISH 的收集操作，還有一種方法可以得到同樣的結果而無需從頭實現新的 Collectors 接口。 Stream 有一個重載的 collect 方法可以接受另外三個函數—— supplier 、accumulator 和 combiner ，其語義和 Collector 接口的相應方法返回的函數完全相同。所以比如說，我們可以像下面這樣把菜肴流中的項目收集到一個 List 中：

List dishes = menuStream.collect(
                ArrayList::new,
                List::add,
                List::addAll);

我們認為，這第二種形式雖然比前一個寫法更為緊湊和簡潔，卻不那么易讀。此外，以恰當的類來實現自己的自定義收集器有助于重用并可避免代碼重復。另外值得注意的是，這第二個collect 方法不能傳遞任何 Characteristics ，所以它永遠都是一個 IDENTITY_FINISH 和CONCURRENT 但并非 UNORDERED 的收集器。

在下一節中，我們一起來實現一個收集器的，讓我們對收集器的新知識更上一層樓。你將會為一個更為復雜，但更為具體、更有說服力的用例開發自己的自定義收集器。

我們用 Collectors 類提供的一個方便的工廠方法創建了一個收集器，它將前n個自然數劃分為質數和非質數，如下所示。

將前n個自然數按質數和非質數分區:

private static Map> partitionPrimes(int n) {
    return IntStream.rangeClosed(2, n).boxed()
            .collect(
                    partitioningBy(candidate -> isPrime(candidate)));
}

當時，通過限制除數不超過被測試數的平方根，我們對最初的 isPrime 方法做了一些改進：

private static boolean isPrime(int candidate) {
    int candidateRoot = (int) Math.sqrt((double) candidate);
    return IntStream.rangeClosed(2, candidateRoot)
            .noneMatch(i -> candidate % i == 0);
}

還有沒有辦法來獲得更好的性能呢？答案是“有”，但為此你必須開發一個自定義收集器。

一個可能的優化是僅僅看看被測試數是不是能夠被質數整除。要是除數本身都不是質數就用不著測了。所以我們可以僅僅用被測試數之前的質數來測試。然而我們目前所見的預定義收集器的問題，也就是必須自己開發一個收集器的原因在于，在收集過程中是沒有辦法訪問部分結果的。這意味著，當測試某一個數字是否是質數的時候，你沒法訪問目前已經找到的其他質數的列表。

假設你有這個列表，那就可以把它傳給 isPrime 方法，將方法重寫如下：

private static boolean isPrime(List primes, int candidate) {
    return primes.stream().noneMatch(i -> candidate % i == 0);
}

而且還應該應用先前的優化，僅僅用小于被測數平方根的質數來測試。因此，你需要想辦法在下一個質數大于被測數平方根時立即停止測試。不幸的是，Stream API中沒有這樣一種方法。你可以使用 filter(p -> p <= candidateRoot) 來篩選出小于被測數平方根的質數。但 filter要處理整個流才能返回恰當的結果。如果質數和非質數的列表都非常大，這就是個問題了。你用不著這樣做；你只需在質數大于被測數平方根的時候停下來就可以了。因此，我們會創建一個名為 takeWhile 的方法，給定一個排序列表和一個謂詞，它會返回元素滿足謂詞的最長前綴：

利用這個方法，你就可以優化 isPrime 方法，只用不大于被測數平方根的質數去測試了：

private static boolean isPrime(List primes, int candidate){
    int candidateRoot = (int) Math.sqrt((double) candidate);
    return takeWhile(primes, i -> i <= candidateRoot)
            .stream()
            .noneMatch(p -> candidate % p == 0);
}

請注意，這個 takeWhile 實現是即時的。理想情況下，我們會想要一個延遲求值的takeWhile ，這樣就可以和 noneMatch 操作合并。不幸的是，這樣的實現超出了本章的范圍，你需要了解Stream API的實現才行。

有了這個新的 isPrime 方法在手，你就可以實現自己的自定義收集器了。首先要聲明一個實現 Collector 接口的新類，然后要開發 Collector 接口所需的五個方法。

1. 第一步：定義 Collector 類的簽名

讓我們從類簽名開始吧，記得 Collector 接口的定義是：

public interface Collector

其中 T 、 A 和 R 分別是流中元素的類型、用于累積部分結果的對象類型，以及 collect 操作最終結果的類型。這里應該收集 Integer 流，而累加器和結果類型則都是 Map>，鍵是 true 和 false ，值則分別是質數和非質數的 List ：

public class PrimeNumbersCollector implements Collector>,
        Map>>

2. 第二步：實現歸約過程

接下來，你需要實現 Collector 接口中聲明的五個方法。 supplier 方法會返回一個在調用時創建累加器的函數：

@Override
public Supplier>> supplier() {
    return () -> new HashMap>(2) {
        {
            put(true, new ArrayList<>());
            put(false, new ArrayList<>());
        }
    };
}

這里不但創建了累積器的Map，還為true和false兩個鍵下面出實話了對應的空列表。在收集過程中會把質數和非指數分別添加到這里。收集器重要的方法是accumulator，因為它定義了如何收集流中元素的邏輯。這里它也是實現了前面所講的優化的關鍵。現在在任何一次迭代中，都可以訪問收集過程的部分結果，也就是包含迄今找到的質數的累加器：

@Override
public BiConsumer>, Integer> accumulator() {
    return ((Map> acc, Integer candidate) -> acc.get(isPrime(acc.get(true), candidate)).add(candidate));
}

在這個個方法中，你調用了isPrime方法，將待測試是否為質數的數以及迄今為止找到的質數列表（也就是累積Map中true鍵對應的值）傳遞給它。這次調用的結果隨后被用作獲取質數或非質數列表的鍵，這樣就可以把新的被測數添加到恰當的列表中。

3.第三步：讓收集器并行工作（如果可能）

下一個方法要在并行收集時把兩個部分累加器合并起來，這里，它只需要合并兩個Map，即將第二個Map中質數和非質數列表中的所有數字合并到第一個Map的對應列表中就行了：

@Override
public BinaryOperator>> combiner() {
    return (Map> map1, Map> map2) -> {
        map1.get(true).addAll(map2.get(true));
        map1.get(false).addAll(map2.get(false));
        return map1;
    };
}

請注意，實際上這個收集器是不能并行的，因為該算法本身是順序的。這意味著永遠都不會調用combiner方法，你可以把它的實現留空。為了讓這個例子完整，我們還是決定實現它。

4.第四步：finisher方法和收集器的characteristics方法

最后兩個方法實現都很簡單。前面說過，accumulator正好就是收集器的結果，也用不著進一步轉換，那么finisher方法就返回identity函數：

@Override
public Function>, Map>> finisher() {
    return Function.identity();
}

就characteristics方法而言，我們已經說過，它既不是CONCURRENT也不是UNOREDERED，但卻是IDENTITY_FINISH的：

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

現在，你可以用這個新的自定義收集器來替代partitioningBy工廠方法創建的那個，并獲得完全相同的結果了：

private static Map> partitionPrimesWithCustomCollector(int n) {
    return IntStream.rangeClosed(2, n).boxed()
            .collect(new PrimeNumbersCollector());
}
Map> primes = partitionPrimesWithCustomCollector(10);
// {false=[4, 6, 8, 9, 10], true=[2, 3, 5, 7]}
System.out.println(primes);

用partitioningBy工廠方法穿件的收集器和你剛剛開發的自定義收集器在功能上是一樣的，但是我們沒有實現用自定義收集器超越partitioningBy收集器性能的目標呢？現在讓我們寫個小程序測試一下吧：

public class CollectorHarness {
    public static void main(String[] args) {
        long fastest = Long.MAX_VALUE;
        // 運行十次
        for (int i = 0; i < 10; i++) {
            long start = System.nanoTime();
            // 將前100萬個自然數按指數和非質數區分
            partitionPrimes(1_000_000);
            long duration = (System.nanoTime() - start) / 1_000_000;
            // 檢查這個執行是否是最快的一個
            if (duration < fastest) {
                fastest = duration;
            }
            System.out.println("done in " + duration);
        }
        System.out.println("Fastest execution done in " + fastest + " msecs");
    }
}

在因特爾I5 6200U 2.4HGz的筆記上運行得到以下的結果：

done in 976
done in 1091
done in 866
done in 867
done in 760
done in 759
done in 777
done in 894
done in 765
done in 763
Fastest execution done in 759 msecs

現在把測試框架的 partitionPrimes 換成 partitionPrimesWithCustomCollector ，以便測試我們開發的自定義收集器的性能。

public class CollectorHarness {
    public static void main(String[] args) {
        excute(PrimeNumbersCollectorExample::partitionPrimesWithCustomCollector);
    }

    private static void excute(Consumer primePartitioner) {
        long fastest = Long.MAX_VALUE;
        // 運行十次
        for (int i = 0; i < 10; i++) {
            long start = System.nanoTime();
            // 將前100萬個自然數按指數和非質數區分
            // partitionPrimes(1_000_000);
            primePartitioner.accept(1_000_000);
            long duration = (System.nanoTime() - start) / 1_000_000;
            // 檢查這個執行是否是最快的一個
            if (duration < fastest) {
                fastest = duration;
            }
            System.out.println("done in " + duration);
        }
        System.out.println("Fastest execution done in " + fastest + " msecs");
    }
}

現在，程序打印：

done in 703
done in 649
done in 715
done in 434
done in 386
done in 403
done in 449
done in 416
done in 353
done in 405
Fastest execution done in 353 msecs

還不錯！看來我們沒有白費功夫開發這個自定義收集器。

collect 是一個終端操作，它接受的參數是將流中元素累積到匯總結果的各種方式（稱為收集器）。

預定義收集器包括將流元素歸約和匯總到一個值，例如計算最小值、最大值或平均值。

預定義收集器可以用 groupingBy 對流中元素進行分組，或用 partitioningBy 進行分區。

收集器可以高效地復合起來，進行多級分組、分區和歸約。

你可以實現 Collector 接口中定義的方法來開發你自己的收集器。

Github:chap6

Gitee:chap6