講透JAVA Stream的collect用法與原理，遠比你想象的更強大

大家好，又見面了，

在我前面的文章《吃透JAVA的Stream流操作，多年實踐總結》中呢，對Stream的整體情況進行了細致全面的講解，也大概介紹了下結果收集器Collectors的常見用法 —— 但遠不是全部，

本篇文章就來專門剖析collect操作，一起解鎖更多高級玩法，讓Stream操作真正的成為我們編碼中的神兵利器，

初識Collector

先看一個簡單的場景：

現有集團內所有人員串列，需要從中篩選出上海子公司的全部人員

假定人員資訊資料如下：

如果你曾經用過Stream流，或者你看過我前面關于Stream用法介紹的文章，那么借助Stream可以很輕松的實作上述訴求：

public void filterEmployeesByCompany() {
    List<Employee> employees = getAllEmployees().stream()
            .filter(employee -> "上海公司".equals(employee.getSubCompany()))
            .collect(Collectors.toList());
    System.out.println(employees);
}

上述代碼中，先創建流，然后通過一系列中間流操作（filter方法）進行業務層面的處理，然后經由終止操作（collect方法）將處理后的結果輸出為List物件，

但我們實際面對的需求場景中，往往會有一些更復雜的訴求，比如說：

現有集團內所有人員串列，需要從中篩選出上海子公司的全部人員，并按照部門進行分組

其實也就是加了個新的分組訴求，那就是先按照前面的代碼實作邏輯基礎上，再對結果進行分組處理就好咯：

public void filterEmployeesThenGroup() {
    // 先 篩選
    List<Employee> employees = getAllEmployees().stream()
            .filter(employee -> "上海公司".equals(employee.getSubCompany()))
            .collect(Collectors.toList());
    // 再 分組
    Map<String, List<Employee>> resultMap = new HashMap<>();
    for (Employee employee : employees) {
        List<Employee> groupList = resultMap
                .computeIfAbsent(employee.getDepartment(), k -> new ArrayList<>());
        groupList.add(employee);
    }
    System.out.println(resultMap);
}

似乎也沒啥毛病，相信很多同學實際編碼中也是這么處理的，但其實我們也可以使用Stream操作直接完成：

public void filterEmployeesThenGroupByStream() {
    Map<String, List<Employee>> resultMap = getAllEmployees().stream()
            .filter(employee -> "上海公司".equals(employee.getSubCompany()))
            .collect(Collectors.groupingBy(Employee::getDepartment));
    System.out.println(resultMap);
}

兩種寫法都可以得到相同的結果：

{
    研發二部=[Employee(subCompany=上海公司, department=研發二部, name=鐵柱, age=34, salary=5000)], 
    研發一部=[Employee(subCompany=上海公司, department=研發一部, name=大壯, age=28, salary=3000), 
             Employee(subCompany=上海公司, department=研發一部, name=二牛, age=24, salary=2000)]
}

上述2種寫法相比而言，第二種是不是代碼上要簡潔很多？而且是不是有種自注釋的味道了？

通過collect方法的合理恰當利用，可以讓Stream適應更多實際的使用場景，大大的提升我們的開發編碼效率，下面就一起來全面認識下collect、解鎖更多高級操作吧，

collect\Collector\Collectors區別與關聯

剛接觸Stream收集器的時候，很多同學都會被collect,Collector,Collectors這幾個概念搞的暈頭轉向，甚至還有很多人即使已經使用Stream好多年，也只是知道collect里面需要傳入類似Collectors.toList()這種簡單的用法，對其背后的細節也不甚了解，

這里以一個collect收集器最簡單的使用場景來剖析說明下其中的關系：

??概括來說：

1?? collect是Stream流的一個終止方法，會使用傳入的收集器（入參）對結果執行相關的操作，這個收集器必須是Collector介面的某個具體實作類
2?? Collector是一個介面，collect方法的收集器是Collector介面的具體實作類
3?? Collectors是一個工具類，提供了很多的靜態工廠方法，提供了很多Collector介面的具體實作類，是為了方便程式員使用而預置的一些較為通用的收集器（如果不使用Collectors類，而是自己去實作Collector介面，也可以），

Collector使用與剖析

到這里我們可以看出，Stream結果收集操作的本質，其實就是將Stream中的元素通過收集器定義的函式處理邏輯進行加工，然后輸出加工后的結果，

根據其執行的操作型別來劃分，又可將收集器分為幾種不同的大類：

下面分別闡述下，

恒等處理Collector

所謂恒等處理，指的就是Stream的元素在經過Collector函式處理前后完全不變，例如toList()操作，只是最終將結果從Stream中取出放入到List物件中，并沒有對元素本身做任何的更改處理：

恒等處理型別的Collector是實際編碼中最常被使用的一種，比如：

list.stream().collect(Collectors.toList());
list.stream().collect(Collectors.toSet());
list.stream().collect(Collectors.toCollection());

歸約匯總Collector

對于歸約匯總類的操作，Stream流中的元素逐個遍歷，進入到Collector處理函式中，然后會與上一個元素的處理結果進行合并處理，并得到一個新的結果，以此類推，直到遍歷完成后，輸出最終的結果，比如Collectors.summingInt()方法的處理邏輯如下：

比如本文開頭舉的例子，如果需要計算上海子公司每個月需要支付的員工總工資，使用Collectors.summingInt()可以這么實作：

public void calculateSum() {
    Integer salarySum = getAllEmployees().stream()
            .filter(employee -> "上海公司".equals(employee.getSubCompany()))
            .collect(Collectors.summingInt(Employee::getSalary));
    System.out.println(salarySum);
}

需要注意的是，這里的匯總計算，不單單只數學層面的累加匯總，而是一個廣義上的匯總概念，即將多個元素進行處理操作，最終生成1個結果的操作，比如計算Stream中最大值的操作，最終也是多個元素中，最終得到一個結果：

還是用之前舉的例子，現在需要知道上海子公司里面工資最高的員工資訊，我們可以這么實作：

public void findHighestSalaryEmployee() {
    Optional<Employee> highestSalaryEmployee = getAllEmployees().stream()
            .filter(employee -> "上海公司".equals(employee.getSubCompany()))
            .collect(Collectors.maxBy(Comparator.comparingInt(Employee::getSalary)));
    System.out.println(highestSalaryEmployee.get());
}

因為這里我們要演示collect的用法，所以用了上述的寫法，實際的時候JDK為了方便使用，也提供了上述邏輯的簡化封裝，我們可以直接使用max()方法來簡化，即上述代碼與下面的寫法等價：

public void findHighestSalaryEmployee2() {
    Optional<Employee> highestSalaryEmployee = getAllEmployees().stream()
            .filter(employee -> "上海公司".equals(employee.getSubCompany()))
            .max(Comparator.comparingInt(Employee::getSalary));
    System.out.println(highestSalaryEmployee.get());
}

分組磁區Collector

Collectors工具類中提供了groupingBy方法用來得到一個分組操作Collector，其內部處理邏輯可以參見下圖的說明：

groupingBy()操作需要指定兩個關鍵輸入，即分組函式和值收集器：

• 分組函式：一個處理函式，用于基于指定的元素進行處理，回傳一個用于分組的值（即分組結果HashMap的Key值），對于經過此函式處理后回傳值相同的元素，將被分配到同一個組里，

• 值收集器：對于分組后的資料元素的進一步處理轉換邏輯，此處還是一個常規的Collector收集器，和collect()方法中傳入的收集器完全等同（可以想想俄羅斯套娃，一個概念），

對于groupingBy分組操作而言，分組函式與值收集器二者必不可少，為了方便使用，在Collectors工具類中，提供了兩個groupingBy多載實作，其中有一個方法只需要傳入一個分組函式即可，這是因為其默認使用了toList()作為值收集器：

例如：僅僅是做一個常規的資料分組操作時，可以僅傳入一個分組函式即可：

public void groupBySubCompany() {
    // 按照子公司維度將員工分組
    Map<String, List<Employee>> resultMap =
            getAllEmployees().stream()
                    .collect(Collectors.groupingBy(Employee::getSubCompany));
    System.out.println(resultMap);
}

這樣collect回傳的結果，就是一個HashMap，其每一個HashValue的值為一個List型別，

而如果不僅需要分組，還需要對分組后的資料進行處理的時候，則需要同時給定分組函式以及值收集器：

public void groupAndCaculate() {
    // 按照子公司分組，并統計每個子公司的員工數
    Map<String, Long> resultMap = getAllEmployees().stream()
            .collect(Collectors.groupingBy(Employee::getSubCompany,
                    Collectors.counting()));
    System.out.println(resultMap);
}

這樣就同時實作了分組與組內資料的處理操作：

{南京公司=2, 上海公司=3}

上面的代碼中Collectors.groupingBy()是一個分組Collector，而其內又傳入了一個歸約匯總Collector Collectors.counting()，也就是一個收集器中嵌套了另一個收集器，

除了上述演示的場景外，還有一種特殊的分組操作，其分組的key型別僅為布林值，這種情況，我們也可以通過Collectors.partitioningBy()提供的磁區收集器來實作，

例如：

統計上海公司和非上海公司的員工總數, true表示是上海公司，false表示非上海公司

使用磁區收集器的方式，可以這么實作：

public void partitionByCompanyAndDepartment() {
    Map<Boolean, Long> resultMap = getAllEmployees().stream()
            .collect(Collectors.partitioningBy(e -> "上海公司".equals(e.getSubCompany()),
                    Collectors.counting()));
    System.out.println(resultMap);
}

結果如下：

{false=2, true=3}

Collectors.partitioningBy()磁區收集器的使用方式與Collectors.groupingBy()分組收集器的使用方式相同，單純從使用維度來看，分組收集器的分組函式回傳值為布林值，則效果等同于一個磁區收集器，

Collector的疊加嵌套

有的時候，我們需要根據先根據某個維度進行分組后，再根據第二維度進一步的分組，然后再對分組后的結果進一步的處理操作，這種場景里面，我們就可以通過Collector收集器的疊加嵌套使用來實作，

例如下面的需求：

現有整個集團全體員工的串列，需要統計各子公司內各部門下的員工人數，

使用Stream的嵌套Collector，我們可以這么實作：

public void groupByCompanyAndDepartment() {
    // 按照子公司+部門雙層維度，統計各個部門內的人員數
    Map<String, Map<String, Long>> resultMap = getAllEmployees().stream()
            .collect(Collectors.groupingBy(Employee::getSubCompany,
                    Collectors.groupingBy(Employee::getDepartment,
                            Collectors.counting())));
    System.out.println(resultMap);
}

可以看下輸出結果，達到了需求預期的訴求：

{
    南京公司={
        測驗二部=1, 
        測驗一部=1}, 
    上海公司={
        研發二部=1, 
        研發一部=2}
}

上面的代碼中,就是一個典型的Collector嵌套處理的例子，同時也是一個典型的多級分組的實作邏輯，對代碼的整體處理程序進行剖析，大致邏輯如下：

借助多個Collector嵌套使用，可以讓我們解鎖很多復雜場景處理能力，你可以將這個操作想象為一個套娃操作，如果愿意，你可以無限嵌套下去（實際中不太可能會有如此荒誕的場景），

Collectors提供的收集器

為了方便程式員使用呢，JDK中的Collectors工具類封裝提供了很多現成的Collector實作類，可供編碼時直接使用，對常用的收集器介紹如下：

上述的大部分方法，前面都有使用示例，這里對collectAndThen補充介紹下，

collectAndThen對應的收集器，必須傳入一個真正用于結果收集處理的實際收集器downstream以及一個finisher方法，當downstream收集器計算出結果后，使用finisher方法對結果進行二次處理，并將處理結果作為最終結果回傳，

還是拿之前的例子來舉例：

給定集團所有員工串列，找出上海公司中工資最高的員工，

我們可以寫出如下代碼：

public void findHighestSalaryEmployee() {
    Optional<Employee> highestSalaryEmployee = getAllEmployees().stream()
            .filter(employee -> "上海公司".equals(employee.getSubCompany()))
            .collect(Collectors.maxBy(Comparator.comparingInt(Employee::getSalary)));
    System.out.println(highestSalaryEmployee.get());
}

但是這個結果最終輸出的是個Optional<Employee>型別，使用的時候比較麻煩，那能不能直接回傳我們需要的Employee型別呢？這里就可以借助collectAndThen來實作：

public void testCollectAndThen() {
    Employee employeeResult = getAllEmployees().stream()
            .filter(employee -> "上海公司".equals(employee.getSubCompany()))
            .collect(
                    Collectors.collectingAndThen(
                            Collectors.maxBy(Comparator.comparingInt(Employee::getSalary)),
                            Optional::get)
            );
    System.out.println(employeeResult);
}

這樣就可以啦，是不是超簡單的？

開發個自定義收集器

前面我們演示了很多Collectors工具類中提供的收集器的用法，上一節中列出來的Collectors提供的常用收集器，也可以覆寫大部分場景的開發訴求了，

但也許在專案中，我們會遇到一些定制化的場景，現有的收集器無法滿足我們的訴求，這個時候，我們也可以自己來實作定制化的收集器，

Collector介面介紹

我們知道，所謂的收集器，其實就是一個Collector介面的具體實作類，所以如果想要定制自己的收集器，首先要先了解Collector介面到底有哪些方法需要我們去實作，以及各個方法的作用與用途，

當我們新建一個MyCollector類并宣告實作Collector介面的時候，會發現需要我們實作5個介面：

這5個介面的含義說明歸納如下：

對于characteristics回傳set集合中的可選值，說明如下：

現在，我們知道了這5個介面方法各自的含義與用途了，那么作為一個Collector收集器，這幾個介面之間是如何配合處理并將Stream資料收集為需要的輸出結果的呢？下面這張圖可以清晰的闡述這一程序：

當然，如果我們的Collector是支持在并行流中使用的，則其處理程序會稍有不同：

為了對上述方法有個直觀的理解，我們可以看下Collectors.toList()這個收集器的實作原始碼：

static final Set<Collector.Characteristics> CH_ID
            = Collections.unmodifiableSet(EnumSet.of(Collector.Characteristics.IDENTITY_FINISH));

public static <T> Collector<T, ?, List<T>> toList() {
    return new CollectorImpl<>((Supplier<List<T>>) ArrayList::new, List::add,
                               (left, right) -> { left.addAll(right); return left; },
                               CH_ID);
}

對上述代碼拆解分析如下：

• supplier方法：ArrayList::new，即new了個ArrayList作為結果存盤容器，

• accumulator方法：List::add，也就是對于stream中的每個元素，都呼叫list.add()方法添加到結果容器追蹤，

• combiner方法：(left, right) -> { left.addAll(right); return left; }，也就是對于并行操作生成的各個子ArrayList結果，最終通過list.addAll()方法合并為最終結果，

• finisher方法：沒提供，使用的默認的，因為無需做任何處理，屬于恒等操作，

• characteristics：回傳的是IDENTITY_FINISH，也即最終結果直接回傳，無需finisher方法去二次加工，注意這里沒有宣告CONCURRENT，因為ArrayList是個非執行緒安全的容器，所以這個收集器是不支持在并發程序中使用，

通過上面的逐個方法描述，再聯想下Collectors.toList()的具體表現，想必對各個介面方法的含義應該有了比較直觀的理解了吧？

實作Collector介面

既然已經搞清楚Collector介面中的主要方法作用，那就可以開始動手寫自己的收集器啦，新建一個class類，然后宣告實作Collector介面，然后去實作具體的介面方法就行咯，

前面介紹過，Collectors.summingInt收集器是用來計算每個元素中某個int型別欄位的總和的，假設我們需要一個新的累加功能：

計算流中每個元素的某個int欄位值平方的總和

下面，我們就一起來自定義一個收集器來實作此功能，

• supplier方法

supplier方法的職責，是創建一個結果存盤累加的容器，既然我們要計算多個值的累加結果，那首先就是要先宣告一個int sum = 0用來存盤累加結果，但是為了讓我們的收集器可以支持在并發模式下使用，我們這里可以采用執行緒安全的AtomicInteger來實作，

所以我們便可以確定supplier方法的實作邏輯了：

@Override
public Supplier<AtomicInteger> supplier() {
    // 指定用于最終結果的收集，此處回傳new AtomicInteger(0)，后續在此基礎上累加
    return () -> new AtomicInteger(0);
}

• accumulator方法

accumulator方法是實作具體的計算邏輯的，也是整個Collector的核心業務邏輯所在的方法，收集器處理的時候，Stream流中的元素會逐個進入到Collector中，然后由accumulator方法來進行逐個計算：

@Override
public BiConsumer<AtomicInteger, T> accumulator() {
    // 每個元素進入的時候的遍歷策略，當前元素值的平方與sum結果進行累加
    return (sum, current) -> {
        int intValue = https://www.cnblogs.com/softwarearch/p/mapper.applyAsInt(current);
        sum.addAndGet(intValue * intValue);
    };
}

這里也補充說下，收集器中的幾個方法中，僅有accumulator是需要重復執行的，有幾個元素就會執行幾次，其余的方法都不會直接與Stream中的元素打交道，

• combiner方法

因為我們前面supplier方法中使用了執行緒安全的AtomicInteger作為結果容器，所以其支持在并行流中使用，根據上面介紹，并行流是將Stream切分為多個分片，然后分別對分片進行計算處理得到分片各自的結果，最后這些分片的結果需要合并為同一份總的結果，這個如何合并，就是此處我們需要實作的：

@Override
public BinaryOperator<AtomicInteger> combiner() {
    // 多個分段結果處理的策略，直接相加
    return (sum1, sum2) -> {
        sum1.addAndGet(sum2.get());
        return sum1;
    };
}

因為我們這里是要做一個數字平方的總和，所以這里對于分片后的結果，我們直接累加到一起即可，

• finisher方法

我們的收集器目標結果是輸出一個累加的Integer結果值，但是為了保證并發流中的執行緒安全，我們使用AtomicInteger作為了結果容器，也就是最終我們需要將內部的AtomicInteger物件轉換為Integer物件，所以finisher方法我們的實作邏輯如下：

@Override
public Function<AtomicInteger, Integer> finisher() {
    // 結果處理完成之后對結果的二次處理
    // 為了支持多執行緒并發處理，此處內部使用了AtomicInteger作為了結果累加器
    // 但是收集器最終需要回傳Integer型別值，此處進行對結果的轉換
    return AtomicInteger::get;
}

• characteristics方法

這里呢，我們宣告下該Collector收集器的一些特性就行了：

1. 因為我們實作的收集器是允許并行流中使用的，所以我們宣告了CONCURRENT屬性；
2. 作為一個數字累加算總和的操作，對元素的先后計算順序并沒有關系，所以我們也同時宣告UNORDERED屬性；
3. 因為我們的finisher方法里面是做了個結果處理轉換操作的，并非是一個恒等處理操作，所以這里就不能宣告IDENTITY_FINISH屬性，

基于此分析，此方法的實作如下：

@Override
public Set<Characteristics> characteristics() {
    Set<Characteristics> characteristics = new HashSet<>();
    // 指定該收集器支持并發處理（前面也發現我們采用了執行緒安全的AtomicInteger方式）
    characteristics.add(Characteristics.CONCURRENT);
    // 宣告元素資料處理的先后順序不影響最終收集的結果
    characteristics.add(Characteristics.UNORDERED);
    // 注意:這里沒有添加下面這句，因為finisher方法對結果進行了處理，非恒等轉換
    // characteristics.add(Characteristics.IDENTITY_FINISH);
    return characteristics;
}

這樣呢，我們的自定義收集器就實作好了，如果需要完整代碼，可以到文末的github倉庫地址上獲取，

我們使用下自己定義的收集器看看：

public void testMyCollector() {
    Integer result = Stream.of(new Score(1), new Score(2), new Score(3), new Score(4))
            .collect(new MyCollector<>(Score::getScore));
    System.out.println(result);
}

輸出結果：

30

完全符合我們的預期，自定義收集器就實作好了，回頭再看下，是不是挺簡單的？

總結

好啦，關于Java中Stream的collect用法與Collector收集器的內容，這里就給大家分享到這里咯，看到這里，不知道你是否掌握了呢？是否還有什么疑問或者更好的見解呢？歡迎多多留言切磋交流，

??此外：

• 關于本文中涉及的演示代碼的完整示例，我已經整理并提交到github中，如果您有需要，可以自取：https://github.com/veezean/JavaBasicSkills

我是悟道，聊技術、又不僅僅聊技術~

如果覺得有用，請點個關注，也可以關注下我的公眾號【架構悟道】，獲取更及時的更新，

期待與你一起探討，一起成長為更好的自己，

本文來自博客園，作者：架構悟道，歡迎關注公眾號[架構悟道]持續獲取更多干貨，轉載請注明原文鏈接：https://www.cnblogs.com/softwarearch/p/16490440.html

標籤：Java

上一篇：資料庫擴容也可以如此絲滑，MySQL千億級資料生產環境擴容實戰

下一篇：02-分布式事務之Seata入門