本列表適用以下定義。

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

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

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

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

public class ToListCollector implements Collector, List>

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

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

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

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

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

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

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

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

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)目的列表：

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

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

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

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

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

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

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

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

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

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

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

5. characteristics 方法

最后一個(gè)方法—— characteristics 會(huì)返回一個(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 是安全的。

我們迄今開發(fā)的 ToListCollector 是 IDENTITY_FINISH 的，因?yàn)橛脕砝鄯e流中元素的List 已經(jīng)是我們要的最終結(jié)果，用不著進(jìn)一步轉(zhuǎn)換了，但它并不是 UNORDERED ，因?yàn)橛迷谟行蛄魃系臅r(shí)候，我們還是希望順序能夠保留在得到的 List 中。最后，它是 CONCURRENT 的，但我們剛才說過了，僅僅在背后的數(shù)據(jù)源無序時(shí)才會(huì)并行處理。

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

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

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

這個(gè)實(shí)現(xiàn)和標(biāo)準(zhǔn)的

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

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

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

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

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

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

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

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

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

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

當(dāng)時(shí)，通過限制除數(shù)不超過被測(cè)試數(shù)的平方根，我們對(duì)最初的 isPrime 方法做了一些改進(jìn)：

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

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

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

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

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

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

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

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

請(qǐng)注意，這個(gè) takeWhile 實(shí)現(xiàn)是即時(shí)的。理想情況下，我們會(huì)想要一個(gè)延遲求值的takeWhile ，這樣就可以和 noneMatch 操作合并。不幸的是，這樣的實(shí)現(xiàn)超出了本章的范圍，你需要了解Stream API的實(shí)現(xiàn)才行。

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

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

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

public interface Collector

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

public class PrimeNumbersCollector implements Collector>,
        Map>>

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

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

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

這里不但創(chuàng)建了累積器的Map，還為true和false兩個(gè)鍵下面出實(shí)話了對(duì)應(yīng)的空列表。在收集過程中會(huì)把質(zhì)數(shù)和非指數(shù)分別添加到這里。收集器重要的方法是accumulator，因?yàn)樗x了如何收集流中元素的邏輯。這里它也是實(shí)現(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));
}

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

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

下一個(gè)方法要在并行收集時(shí)把兩個(gè)部分累加器合并起來，這里，它只需要合并兩個(gè)Map，即將第二個(gè)Map中質(zhì)數(shù)和非質(zhì)數(shù)列表中的所有數(shù)字合并到第一個(gè)Map的對(duì)應(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;
    };
}

請(qǐng)注意，實(shí)際上這個(gè)收集器是不能并行的，因?yàn)樵撍惴ū旧硎琼樞虻?。這意味著永遠(yuǎn)都不會(huì)調(diào)用combiner方法，你可以把它的實(shí)現(xiàn)留空。為了讓這個(gè)例子完整，我們還是決定實(shí)現(xiàn)它。

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

最后兩個(gè)方法實(shí)現(xiàn)都很簡(jiǎn)單。前面說過，accumulator正好就是收集器的結(jié)果，也用不著進(jìn)一步轉(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)在，你可以用這個(gè)新的自定義收集器來替代partitioningBy工廠方法創(chuàng)建的那個(gè)，并獲得完全相同的結(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ā)的自定義收集器在功能上是一樣的，但是我們沒有實(shí)現(xiàn)用自定義收集器超越partitioningBy收集器性能的目標(biāo)呢？現(xiàn)在讓我們寫個(gè)小程序測(cè)試一下吧：

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

在因特爾I5 6200U 2.4HGz的筆記上運(yùn)行得到以下的結(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)在把測(cè)試框架的 partitionPrimes 換成 partitionPrimesWithCustomCollector ，以便測(cè)試我們開發(fā)的自定義收集器的性能。

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

    private static void excute(Consumer primePartitioner) {
        long fastest = Long.MAX_VALUE;
        // 運(yùn)行十次
        for (int i = 0; i < 10; i++) {
            long start = System.nanoTime();
            // 將前100萬個(gè)自然數(shù)按指數(shù)和非質(zhì)數(shù)區(qū)分
            // partitionPrimes(1_000_000);
            primePartitioner.accept(1_000_000);
            long duration = (System.nanoTime() - start) / 1_000_000;
            // 檢查這個(gè)執(zhí)行是否是最快的一個(gè)
            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

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

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

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

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

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

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

Github:chap6

Gitee:chap6