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

語言: CN / TW / HK
時間 2022-07-19 03:15:49

theme: healer-readable

大家好，又見面了。

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

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

初識Collector

先看一個簡單的場景：

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

假定人員資訊資料如下：

| 姓名 | 子公司 | 部門 | 年齡 | 工資 | | ---- | -------- | -------- | ---- | ---- | | 大壯 | 上海公司 | 研發一部 | 28 | 3000 | | 二牛 | 上海公司 | 研發一部 | 24 | 2000 | | 鐵柱 | 上海公司 | 研發二部 | 34 | 5000 | | 翠花 | 南京公司 | 測試一部 | 27 | 3000 | | 玲玲 | 南京公司 | 測試二部 | 31 | 4000 |

如果你曾經用過Stream流，或者你看過我前面關於Stream用法介紹的文章，那麼藉助Stream可以很輕鬆的實現上述訴求：

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

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

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

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

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

java 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操作直接完成：

```java public void filterEmployeesThenGroupByStream() { Map> 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是實際編碼中最常被使用的一種，比如：

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

歸約彙總Collector

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

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

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

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

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

java 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()方法來簡化，即上述程式碼與下面的寫法等價：

java 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()作為值收集器：

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

java public void groupBySubCompany() { // 按照子公司維度將員工分組 Map<String, List<Employee>> resultMap = getAllEmployees().stream() .collect(Collectors.groupingBy(Employee::getSubCompany)); System.out.println(resultMap); } 這樣collect返回的結果，就是一個HashMap，其每一個HashValue的值為一個List型別。

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

java 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表示非上海公司

使用分割槽收集器的方式，可以這麼實現：

java 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，我們可以這麼實現：

java 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實現類，可供編碼時直接使用，對常用的收集器介紹如下：

| 方法 | 含義說明 | | ----------------- | ------------------------------------------------------------------------------------------ | | toList | 將流中的元素收集到一個List中 | | toSet | 將流中的元素收集到一個Set中 | | toCollection | 將流中的元素收集到一個Collection中 | | toMap | 將流中的元素對映收集到一個Map中 | | counting | 統計流中的元素個數 | | summingInt | 計算流中指定int欄位的累加總和。針對不同型別的數字型別，有不同的方法，比如summingDouble等 | | averagingInt | 計算流中指定int欄位的平均值。針對不同型別的數字型別，有不同的方法，比如averagingLong等 | | joining | 將流中所有元素（或者元素的指定欄位）字串值進行拼接，可以指定拼接連線符，或者首尾拼接字元 | | maxBy | 根據給定的比較器，選擇出值最大的元素 | | minBy | 根據給定的比較器，選擇出值最小的元素 | | groupingBy | 根據給定的分組函式的值進行分組，輸出一個Map物件 | | partitioningBy | 根據給定的分割槽函式的值進行分割槽，輸出一個Map物件，且key始終為布林值型別 | | collectingAndThen | 包裹另一個收集器，對其結果進行二次加工轉換 | | reducing | 從給定的初始值開始，將元素進行逐個的處理，最終將所有元素計算為最終的1個值輸出 |

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

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

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

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

我們可以寫出如下程式碼：

java 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來實現：

java 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個介面的含義說明歸納如下：

| 介面名稱 | 功能含義說明 | |---|---| | supplier | 建立新的結果容器，可以是一個容器，也可以是一個累加器例項，總之是用來儲存結果資料的 | | accumlator | 元素進入收集器中的具體處理操作 | | finisher | 當所有元素都處理完成後，在返回結果前的對結果的最終處理操作，當然也可以選擇不做任何處理，直接返回 | | combiner | 各個子流的處理結果最終如何合併到一起去，比如並行流處理場景，元素會被切分為好多個分片進行並行處理，最終各個分片的資料需要合併為一個整體結果，即通過此方法來指定子結果的合併邏輯 | | characteristics | 對此收集器處理行為的補充描述，比如此收集器是否允許並行流中處理，是否finisher方法必須要有等等，此處返回一個Set集合，裡面的候選值是固定的幾個可選項。 |

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

| 取值 | 含義說明 | |---|---| | UNORDERED | 宣告此收集器的彙總歸約結果與Stream流元素遍歷順序無關，不受元素處理順序影響 | | CONCURRENT | 宣告此收集器可以多個執行緒並行處理，允許並行流中進行處理 | | IDENTITY_FINISH | 宣告此收集器的finisher方法是一個恆等操作，可以跳過 |

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

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

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

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

public static Collector> toList() { return new CollectorImpl<>((Supplier>) 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方法的實現邏輯了：

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

  • accumulator方法

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

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

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

  • combiner方法

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

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

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

  • finisher方法

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

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

  • characteristics方法

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

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

基於此分析，此方法的實現如下：

java @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倉庫地址上獲取。

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

java 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收集器的內容，這裡就給大家分享到這裡咯。看到這裡，不知道你是否掌握了呢？是否還有什麼疑問或者更好的見解呢？歡迎多多留言切磋交流。

📢此外：

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

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

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

我正在參與掘金技術社群創作者簽約計劃招募活動，點選連結報名投稿。

「」
「」