Pre

Collector 接口包含了一系列方法，为实现具体的归约操作（即收集器）提供了范本。

我们已经看过了 Collector 接口中实现的许多收集器，例如 toList 或 groupingBy 。这也意味着可以为 Collector 接口提供自己的实现，从而自由地创建自定义归约操作。

要开始使用 Collector 接口，我们先看看toList 工厂方法，它会把流中的所有元素收集成一个 List 。我们当时说在日常工作中经常会用到这个收集器，而且它也是写起来比较直观的一个，至少理论上如此。通过仔细研究这个收集器是怎么实现的。

我们可以很好地了解 Collector 接口是怎么定义的，以及它的方法所返回的函数在内部是如何为collect 方法所用的。

Collector接口声明的方法

首先让我们在下面的列表中看看 Collector 接口的定义，它列出了接口的签名以及声明的五个方法。

• T 是流中要收集的项目的泛型。
• A 是累加器的类型，累加器是在收集过程中用于累积部分结果的对象。
• R 是收集操作得到的对象（通常但并不一定是集合）的类型。

例如，你可以实现一个 ToListCollector<T> 类，将 Stream<T> 中的所有元素收集到一个List<T> 里，它的签名如下

public class ToListCollector<T> implements Collector<T, List<T>, List<T>>

待会揭秘 这里用于累积的对象也将是收集过程的最终结果

理解 Collector接口中声明的方法

现在我们可以一个个来分析Collector 接口声明的五个方法了。通过分析，你会注意到，前四个方法都会返回一个会被 collect 方法调用的函数，而第五个方法 characteristics 则提供了一系列特征，也就是一个提示列表，告诉 collect 方法在执行归约操作的时候可以应用哪些优化（比如并行化）

1.建立新的结果容器： supplier 方法

supplier 方法必须返回一个结果为空的 Supplier ，也就是一个无参数函数，在调用时它会创建一个空的累加器实例，供数据收集过程使用。

很明显，对于将累加器本身作为结果返回的收集器，比如我们的 ToListCollector ，在对空流执行操作的时候，这个空的累加器也代表了收集过程的结果。

在我们的 ToListCollector 中， supplier 返回一个空的 List ，如下所示：

public Supplier<List<T>> supplier() {
  return () -> new ArrayList<T>();
}

请注意你也可以只传递一个构造函数引用：

public Supplier<List<T>> supplier() {
return ArrayList::new;
}

2.将元素添加到结果容器： accumulator 方法

accumulator 方法会返回执行归约操作的函数。当遍历到流中第n个元素时，这个函数执行时会有两个参数：保存归约结果的累加器（已收集了流中的前 n-1 个项目），还有第n个元素本身。

该函数将返回 void ，因为累加器是原位更新，即函数的执行改变了它的内部状态以体现遍历的元素的效果。

对于 ToListCollector ，这个函数仅仅会把当前项目添加至已经遍历过的项目的列表：

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

你也可以使用方法引用，这会更为简洁：

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

3.对结果容器应用最终转换： finisher 方法

在遍历完流后， finisher 方法必须返回在累积过程的最后要调用的一个函数，以便将累加器对象转换为整个集合操作的最终结果。

通常，就像 ToListCollector 的情况一样，累加器对象恰好符合预期的最终结果，因此无需进行转换。所以 finisher 方法只需返回 identity 函数：

public Function<List<T>, List<T>> finisher() {
  return Function.identity();
}

这三个方法已经足以对流进行顺序归约，至少从逻辑上看可以按下图进行。

实践中的实现细节可能还要复杂一点，一方面是因为流的延迟性质，可能在 collect 操作之前还需要完成其他中间操作的流水线，另一方面则是理论上可能要进行并行归约。

4.合并两个结果容器： combiner 方法

四个方法中的最后一个—— combiner 方法会返回一个供归约操作使用的函数，它定义了对

流的各个子部分进行并行处理时，各个子部分归约所得的累加器要如何合并。

对于 toList 而言，这个方法的实现非常简单，只要把从流的第二个部分收集到的项目列表加到遍历第一部分时得到的列表后面就行了：

public BinaryOperator<List<T>> combiner() {
  return (list1, list2) -> {
    list1.addAll(list2);
    return list1; 
  }
}

有了这第四个方法，就可以对流进行并行归约了。它会用到Java 7中引入的分支/合并框架和Spliterator 抽象， 如下图所示

原始流会以递归方式拆分为子流，直到定义流是否需要进一步拆分的一个条件为非（如果分布式工作单位太小，并行计算往往比顺序计算要慢，而且要是生成的并行任务比处理器内核数多很多的话就毫无意义了）。

现在，所有的子流都可以并行处理，即对每个子流应用【见顺序归约过程的逻辑步骤】的顺序归约算法。

最后，使用收集器 combiner 方法返回的函数，将所有的部分结果两两合并。这时会把原始流每次拆分时得到的子流对应的结果合并起来。

5. characteristics 方法

最后一个方法—— characteristics 会返回一个不可变的 Characteristics 集合，它定义了收集器的行为——尤其是关于流是否可以并行归约，以及可以使用哪些优化的提示。

Characteristics 是一个包含三个项目的枚举。

UNORDERED ——归约结果不受流中项目的遍历和累积顺序的影响

CONCURRENT —— accumulator 函数可以从多个线程同时调用，且该收集器可以并行归约流。如果收集器没有标为 UNORDERED ，那它仅在用于无序数据源时才可以并行归约。

IDENTITY_FINISH ——这表明完成器方法返回的函数是一个恒等函数，可以跳过。这种情况下，累加器对象将会直接用作归约过程的最终结果。这也意味着，将累加器 A 不加检查地转换为结果 R 是安全的。

我们迄今开发的 ToListCollector 是 IDENTITY_FINISH 的，因为用来累积流中元素的List 已经是我们要的最终结果，用不着进一步转换了，但它并不是 UNORDERED ，因为用在有序流上的时候，我们还是希望顺序能够保留在得到的 List 中。

最后，它是CONCURRENT 的，但我们刚才说过了，仅仅在背后的数据源无序时才会并行处理。

自定义Collector Demo

public class ToListCollector<T> implements Collector<T, List<T>, List<T>> {
    private void log(final String log) {
        System.out.println(Thread.currentThread().getName() + "-" + log);
    }
    @Override
    public Supplier<List<T>> supplier() {
        log("supplier");
        return ArrayList::new;
    }
    @Override
    public BiConsumer<List<T>, T> accumulator() {
        log("accumulator");
        return List::add;
    }
    @Override
    public BinaryOperator<List<T>> combiner() {
        log("combiner");
        return (list1, list2) -> {
            list1.addAll(list2);
            return list1;
        };
    }
    @Override
    public Function<List<T>, List<T>> finisher() {
        log("finisher");
        return t -> t;
    }
    @Override
    public Set<Characteristics> characteristics() {
        log("characteristics");
        return Collections.unmodifiableSet(
                EnumSet.of(Characteristics.IDENTITY_FINISH, Characteristics.CONCURRENT
                ));
    }

测试下

public class CustomerCollectorAction {
    public static void main(String[] args) {
        Collector<String, List<String>, List<String>> collector = new ToListCollector<>();
        String[] arrs = new String[]{"Artisan", "Yang", "Hello", "Lambda", "Collector", "Java 8", "Stream"};
        List<String> result = Arrays.asList( arrs)
                .parallelStream()
                .filter(s -> s.length() >= 5)
                .collect(collector);
        System.out.println(result);
    }
}