本列表適用以下定義。

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

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

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

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

public class ToListCollector implements Collector, List>

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

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

1. 建立新的結(jié)果容器： supplier 方法

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

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

請注意你也可以只傳遞一個構(gòu)造函數(shù)引用：

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

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

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

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

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

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

3. 對結(jié)果容器應(yīng)用最終轉(zhuǎn)換： finisher 方法

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

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

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

4. 合并兩個結(jié)果容器： combiner 方法

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

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

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

5. characteristics 方法

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

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

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

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

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

前一小節(jié)中談到的五個方法足夠我們開發(fā)自己的 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));
    }
}

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

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

這個實現(xiàn)和標準的

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

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

進行自定義收集而不去實現(xiàn) Collector

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

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

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

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

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

將前n個自然數(shù)按質(zhì)數(shù)和非質(zhì)數(shù)分區(qū):

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

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

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

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

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

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

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

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

利用這個方法，你就可以優(yōu)化 isPrime 方法，只用不大于被測數(shù)平方根的質(zhì)數(shù)去測試了：

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 實現(xiàn)是即時的。理想情況下，我們會想要一個延遲求值的takeWhile ，這樣就可以和 noneMatch 操作合并。不幸的是，這樣的實現(xiàn)超出了本章的范圍，你需要了解Stream API的實現(xiàn)才行。

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

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

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

public interface Collector

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

public class PrimeNumbersCollector implements Collector>,
        Map>>

2. 第二步：實現(xiàn)歸約過程

接下來，你需要實現(xiàn) Collector 接口中聲明的五個方法。 supplier 方法會返回一個在調(diào)用時創(chuàng)建累加器的函數(shù)：

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

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

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

在這個個方法中，你調(diào)用了isPrime方法，將待測試是否為質(zhì)數(shù)的數(shù)以及迄今為止找到的質(zhì)數(shù)列表（也就是累積Map中true鍵對應(yīng)的值）傳遞給它。這次調(diào)用的結(jié)果隨后被用作獲取質(zhì)數(shù)或非質(zhì)數(shù)列表的鍵，這樣就可以把新的被測數(shù)添加到恰當?shù)牧斜碇小?/p>

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

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

@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;
    };
}

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

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

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

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

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

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

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

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工廠方法穿件的收集器和你剛剛開發(fā)的自定義收集器在功能上是一樣的，但是我們沒有實現(xiàn)用自定義收集器超越partitioningBy收集器性能的目標呢？現(xiàn)在讓我們寫個小程序測試一下吧：

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萬個自然數(shù)按指數(shù)和非質(zhì)數(shù)區(qū)分
            partitionPrimes(1_000_000);
            long duration = (System.nanoTime() - start) / 1_000_000;
            // 檢查這個執(zhí)行是否是最快的一個
            if (duration < fastest) {
                fastest = duration;
            }
            System.out.println("done in " + duration);
        }
        System.out.println("Fastest execution done in " + fastest + " msecs");
    }
}

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

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

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

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萬個自然數(shù)按指數(shù)和非質(zhì)數(shù)區(qū)分
            // partitionPrimes(1_000_000);
            primePartitioner.accept(1_000_000);
            long duration = (System.nanoTime() - start) / 1_000_000;
            // 檢查這個執(zhí)行是否是最快的一個
            if (duration < fastest) {
                fastest = duration;
            }
            System.out.println("done in " + duration);
        }
        System.out.println("Fastest execution done in " + fastest + " msecs");
    }
}

現(xiàn)在，程序打?。?/p>

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

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

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

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

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

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

你可以實現(xiàn) Collector 接口中定義的方法來開發(fā)你自己的收集器。

Github:chap6

Gitee:chap6