不吹不黑 JAVA Stream 的 collect 用法与原理，远比你想象的更强大

初识 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 对象。

​

编辑切换为居中

添加图片注释，不超过 140 字（可选）

但我们实际面对的需求场景中，往往会有一些更复杂的诉求，比如说：

现有集团内所有人员列表，需要从中筛选出上海子公司的全部人员，并按照部门进行分组

其实也就是加了个新的分组诉求，那就是先按照前面的代码实现逻辑基础上，再对结果进行分组处理就好咯：

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 收集器最简单的使用场景来剖析说明下其中的关系：

​

编辑切换为居中

添加图片注释，不超过 140 字（可选）

📢概括来说：

1️⃣ collect 是 Stream 流的一个终止方法，会使用传入的收集器（入参）对结果执行相关的操作，这个收集器必须是 Collector 接口的某个具体实现类 2️⃣ Collector 是一个接口，collect 方法的收集器是 Collector 接口的具体实现类 3️⃣ Collectors 是一个工具类，提供了很多的静态工厂方法，提供了很多 Collector 接口的具体实现类，是为了方便程序员使用而预置的一些较为通用的收集器（如果不使用 Collectors 类，而是自己去实现 Collector 接口，也可以）。

Collector 使用与剖析

到这里我们可以看出，Stream 结果收集操作的本质，其实就是将 Stream 中的元素通过收集器定义的函数处理逻辑进行加工，然后输出加工后的结果。

​

编辑切换为居中

添加图片注释，不超过 140 字（可选）

根据其执行的操作类型来划分，又可将收集器分为几种不同的大类：

​

编辑切换为居中

添加图片注释，不超过 140 字（可选）

下面分别阐述下。

恒等处理 Collector

所谓恒等处理，指的就是 Stream 的元素在经过 Collector 函数处理前后完全不变，例如 toList()操作，只是最终将结果从 Stream 中取出放入到 List 对象中，并没有对元素本身做任何的更改处理：

​

编辑切换为居中

添加图片注释，不超过 140 字（可选）

恒等处理类型的 Collector 是实际编码中最常被使用的一种，比如：

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

归约汇总 Collector

对于归约汇总类的操作，Stream 流中的元素逐个遍历，进入到 Collector 处理函数中，然后会与上一个元素的处理结果进行合并处理，并得到一个新的结果，以此类推，直到遍历完成后，输出最终的结果。比如 Collectors.summingInt()方法的处理逻辑如下：

​

编辑切换为居中

添加图片注释，不超过 140 字（可选）

比如本文开头举的例子，如果需要计算上海子公司每个月需要支付的员工总工资，使用 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 中最大值的操作，最终也是多个元素中，最终得到一个结果：

​

编辑切换为居中

添加图片注释，不超过 140 字（可选）

还是用之前举的例子，现在需要知道上海子公司里面工资最高的员工信息，我们可以这么实现：

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，其内部处理逻辑可以参见下图的说明：

​

编辑切换为居中

添加图片注释，不超过 140 字（可选）

groupingBy()操作需要指定两个关键输入，即分组函数和值收集器：

• 分组函数：一个处理函数，用于基于指定的元素进行处理，返回一个用于分组的值（即分组结果 HashMap 的 Key 值），对于经过此函数处理后返回值相同的元素，将被分配到同一个组里。

• 值收集器：对于分组后的数据元素的进一步处理转换逻辑，此处还是一个常规的 Collector 收集器，和 collect()方法中传入的收集器完全等同（可以想想俄罗斯套娃，一个概念）。

对于 groupingBy 分组操作而言，分组函数与值收集器二者必不可少。为了方便使用，在 Collectors 工具类中，提供了两个 groupingBy 重载实现，其中有一个方法只需要传入一个分组函数即可，这是因为其默认使用了 toList()作为值收集器：

​

编辑切换为居中

添加图片注释，不超过 140 字（可选）

例如：仅仅是做一个常规的数据分组操作时，可以仅传入一个分组函数即可：

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 嵌套处理的例子，同时也是一个典型的多级分组的实现逻辑。对代码的整体处理过程进行剖析，大致逻辑如下：

​

编辑切换为居中

添加图片注释，不超过 140 字（可选）

借助多个 Collector 嵌套使用，可以让我们解锁很多复杂场景处理能力。你可以将这个操作想象为一个套娃操作，如果愿意，你可以无限嵌套下去（实际中不太可能会有如此荒诞的场景）。

Collectors 提供的收集器

为了方便程序员使用呢，JDK 中的 Collectors 工具类封装提供了很多现成的 Collector 实现类，可供编码时直接使用，对常用的收集器介绍如下：

上述的大部分方法，前面都有使用示例，这里对 collectAndThen 补充介绍下。

collectAndThen 对应的收集器，必须传入一个真正用于结果收集处理的实际收集器 downstream 以及一个 finisher 方法，当 downstream 收集器计算出结果后，使用 finisher 方法对结果进行二次处理，并将处理结果作为最终结果返回。

​

编辑切换为居中

添加图片注释，不超过 140 字（可选）

还是拿之前的例子来举例：

给定集团所有员工列表，找出上海公司中工资最高的员工。

我们可以写出如下代码：

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 个接口：

​

编辑

添加图片注释，不超过 140 字（可选）

这 5 个接口的含义说明归纳如下：

对于 characteristics 返回 set 集合中的可选值，说明如下：

现在，我们知道了这 5 个接口方法各自的含义与用途了，那么作为一个 Collector 收集器，这几个接口之间是如何配合处理并将 Stream 数据收集为需要的输出结果的呢？下面这张图可以清晰的阐述这一过程：

​

编辑切换为居中

添加图片注释，不超过 140 字（可选）

当然，如果我们的 Collector 是支持在并行流中使用的，则其处理过程会稍有不同：

​

编辑切换为居中

添加图片注释，不超过 140 字（可选）

为了对上述方法有个直观的理解，我们可以看下 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 方法的实现逻辑了：

@Overridepublic Supplier<AtomicInteger> supplier() {    // 指定用于最终结果的收集，此处返回new AtomicInteger(0)，后续在此基础上累加    return () -> new AtomicInteger(0);}

• accumulator 方法

accumulator 方法是实现具体的计算逻辑的，也是整个 Collector 的核心业务逻辑所在的方法。收集器处理的时候，Stream 流中的元素会逐个进入到 Collector 中，然后由 accumulator 方法来进行逐个计算：

@Overridepublic BiConsumer<AtomicInteger, T> accumulator() {    // 每个元素进入的时候的遍历策略，当前元素值的平方与sum结果进行累加    return (sum, current) -> {        int intValue = mapper.applyAsInt(current);        sum.addAndGet(intValue * intValue);    };}

这里也补充说下，收集器中的几个方法中，仅有 accumulator 是需要重复执行的，有几个元素就会执行几次，其余的方法都不会直接与 Stream 中的元素打交道。

• combiner 方法

因为我们前面 supplier 方法中使用了线程安全的 AtomicInteger 作为结果容器，所以其支持在并行流中使用。根据上面介绍，并行流是将 Stream 切分为多个分片，然后分别对分片进行计算处理得到分片各自的结果，最后这些分片的结果需要合并为同一份总的结果，这个如何合并，就是此处我们需要实现的：

@Overridepublic BinaryOperator<AtomicInteger> combiner() {    // 多个分段结果处理的策略，直接相加    return (sum1, sum2) -> {        sum1.addAndGet(sum2.get());        return sum1;    };}

因为我们这里是要做一个数字平方的总和，所以这里对于分片后的结果，我们直接累加到一起即可。

• finisher 方法

我们的收集器目标结果是输出一个累加的 Integer 结果值，但是为了保证并发流中的线程安全，我们使用 AtomicInteger 作为了结果容器。也就是最终我们需要将内部的 AtomicInteger 对象转换为 Integer 对象，所以 finisher 方法我们的实现逻辑如下：

@Overridepublic Function<AtomicInteger, Integer> finisher() {    // 结果处理完成之后对结果的二次处理    // 为了支持多线程并发处理，此处内部使用了AtomicInteger作为了结果累加器    // 但是收集器最终需要返回Integer类型值，此处进行对结果的转换    return AtomicInteger::get;}

• characteristics 方法

这里呢，我们声明下该 Collector 收集器的一些特性就行了：

1. 因为我们实现的收集器是允许并行流中使用的，所以我们声明了 CONCURRENT 属性；

2. 作为一个数字累加算总和的操作，对元素的先后计算顺序并没有关系，所以我们也同时声明 UNORDERED 属性；

3. 因为我们的 finisher 方法里面是做了个结果处理转换操作的，并非是一个恒等处理操作，所以这里就不能声明 IDENTITY_FINISH 属性。

基于此分析，此方法的实现如下：

@Overridepublic 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 收集器的内容，这里就给大家分享到这里咯。看到这里，不知道你是否掌握了呢？是否还有什么疑问或者更好的见解呢？欢迎多多留言切磋交流。

原文链接：讲透JAVA Stream的collect用法与原理，远比你想象的更强大