(2) Java 8 实战第二版——补充 收集数据、并行数据处理能力与性能

第6章

• 用Collectors类创建和使用收集器
• 将数据流归约为一个值
• 汇总：归约的特殊情况
• 数据分组和分区
• 开发你的自定义收集器

对一个交易列表按货币分组，获得该货币的所有交易额总和（返回一个Map）。
将交易列表分成两组：贵的和不贵的（返回一个Map）。
创建多级分组，比如按城市对交易分组，然后进一步按照贵或不贵分组（返回一个Map>）。
有一个由Transaction构成的List，并且想按照名义货币进行分组。在Java 8之前

Map<Currency, List<Transaction>> transactionsByCurrencies = new HashMap<>();      
←---- 建立累积交易分组的Map
for (Transaction transaction : transactions) {       ←---- 迭代Transaction的List
    Currency currency = transaction.getCurrency();       ←---- 提取Transaction的货币
    List<Transaction> transactionsForCurrency = transactionsByCurrencies.get(currency);
    if (transactionsForCurrency == null) {       ←---- 如果分组Map中没有这种货币的条目，就创建一个
        transactionsForCurrency = new ArrayList<>();
        transactionsByCurrencies
                            .put(currency, transactionsForCurrency);
    }
    transactionsForCurrency.add(transaction);       ←---- 将当前遍历的Transaction加入同一货币的Transaction的List
}
1
2
3
4
5
6
7
8
9
10
11
12

Map<Currency, List<Transaction>> transactionsByCurrencies =
        transactions.stream().collect(groupingBy(Transaction::getCurrency));
1
2

预定义收集器

1. 将流元素归约和汇总为一个值；
2. 元素分组；
3. 元素分区。
   

规约

long howManyDishes = menu.stream().collect(Collectors.counting());
//等同于
long howManyDishes = menu.stream().count();

1
2
3
4

查找最大最小

可以使用两个收集器，Collectors.maxBy和Collectors.minBy，来计算流中的最大值或最小值

Comparator<Dish> dishCaloriesComparator =
    Comparator.comparingInt(Dish::getCalories);
Optional<Dish> mostCalorieDish =
    menu.stream()
        .collect(maxBy(dishCaloriesComparator));
1
2
3
4
5

汇总

int totalCalories = menu.stream().collect(summingInt(Dish::getCalories));
//遍历流时，会把每一道菜都映射为其热量，然后把这个数字累加到一个累加器（这里的初始值0）
1
2

Collectors.summingLong和Collectors.summingDouble方法的作用完全一样，可以用于求和字段为long或double的情况。
汇总不仅仅是求和；还有Collectors.averagingInt，连同对应的averagingLong和averagingDouble可以计算数值的平均数

double avgCalories =
    menu.stream().collect(averagingInt(Dish::getCalories));
1
2

使用summarizingInt工厂方法返回的收集器。例如，通过一次summarizing操作你就可以数出菜单中元素的个数，并得到菜肴热量总和、平均值、最大值和最小值

IntSummaryStatistics menuStatistics =
        menu.stream().collect(summarizingInt(Dish::getCalories));
//这个收集器会把所有这些信息收集到一个叫作IntSummaryStatistics的类里，
//它提供了方便的取值（getter）方法来访问结果。打印menuStatisticobject会得到以下输出：
IntSummaryStatistics{count=9, sum=4300, min=120,
                     average=477.777778, max=800}
1
2
3
4
5
6

同样，相应的summarizingLong和summarizingDouble工厂方法有相关的LongSummary Statistics和DoubleSummaryStatistics类型，适用于收集的属性是原始类型long或double的情况。

连接字符串

//joining工厂方法返回的收集器会把对流中每一个对象应用toString方法
//得到的所有字符串连接成一个字符串
String shortMenu = menu.stream().map(Dish::getName).collect(joining());
1
2
3

注意，joining在内部使用了StringBuilder来把生成的字符串逐个追加起来。
此外还要注意，如果Dish类有一个toString方法来返回菜肴的名称，则无需用提取每一道菜名称的函数来对原流做映射就能够得到相同的结果：

String shortMenu = menu.stream().collect(joining());
但该字符串的可读性并不好。
幸好，joining工厂方法有一个重载版本可以接受元素之间的分界符，
这样你就可以得到一个逗号分隔的菜肴名称列表：
String shortMenu = menu.stream().map(Dish::getName).collect(joining(", "));
1
2
3
4
5

二者均可产生以下字符串：（无逗号）
porkbeefchickenfrench friesriceseason fruitpizzaprawnssalmon
二者均可产生以下字符串：（有逗号）
pork, beef, chicken, french fries, rice, season fruit, pizza, prawns, salmon

广义的归约汇总

事实上，大部分的收集器，都是一个可以用reducing工厂方法定义的归约过程的特殊情况而已。
Collectors.reducing工厂方法是所有这些特殊情况的一般化。
可以说，大部分常用的案例仅仅是为了方便程序员而已。
（但是，请记得方便程序员和可读性是头等大事！）
例如，可以用reducing方法创建的收集器来计算你菜单的总热量，如下所示：

int totalCalories = menu.stream().collect(reducing(
                                   0, Dish::getCalories, (i, j) -> i + j));
1
2

它需要三个参数。

1. 第一个参数是归约操作的起始值，也是流中没有元素时的返回值，所以很显然对于数值和而言0是一个合适的值。
2. 第二个参数reduce函数，将菜肴转换成一个表示其所含热量的int。
3. 第三个参数是一个BinaryOperator，将两个项目累积成一个同类型的值。这里它就是对两个int求和。

Optional<Dish> mostCalorieDish =
    menu.stream().collect(reducing(
        (d1, d2) -> d1.getCalories() > d2.getCalories() ? d1 : d2));
1
2
3

单参数reducing工厂方法创建的收集器看作三参数方法的特殊情况，把流中的第一个项目作为起点，把恒等函数（即一个函数仅仅是返回其输入参数）作为一个转换函数。
意味着，要是把单参数reducing收集器传递给空流的collect方法，收集器就没有起点；它将因此而返回一个Optional对象。

收集和规约

Stream接口的collect和reduce方法有何不同，因为两种方法通常会获得相同的结果

Stream<Integer> stream = Arrays.asList(1, 2, 3, 4, 5, 6).stream();
List<Integer> numbers = stream.reduce(
                               new ArrayList<Integer>(),
                               (List<Integer> l, Integer e) -> {
                                         l.add(e);
                                         return l; },
                               (List<Integer> l1, List<Integer> l2) -> {
                                         l1.addAll(l2);
                                         return l1; });

这个归约过程不能并行工作，因为由多个线程并发修改同一个数据结构可能会破坏List本身。
在这种情况下，如果你想要线程安全，就需要每次分配一个新的List，而对象分配又会影响性能。
这就是collect方法特别适合表达可变容器上的归约的原因，更关键的是它适合并行操作
1
2
3
4
5
6
7
8
9
10
11
12
13

这个解决方案有两个问题：一个语义问题和一个实际问题。语义问题在于，reduce方法旨在把两个值结合起来生成一个新值，它是一个不可变的归约。
与此相反，collect方法的设计就是要改变容器，从而累积要输出的结果。这意味着，上面的代码片段是在滥用reduce方法，因为它在原地改变了作为累加器的List。

int totalCalories = menu.stream().collect(reducing(0,       ←---- 初始值
                              Dish::getCalories,       ←---- 转换函数
                              Integer::sum));       ←---- 累积函数
//利用累积函数，把一个初始化为起始值的累加器，
//和把转换函数应用到流中每个元素上得到的结果不断迭代合并起来
1
2
3
4
5

counting收集器也是类似地利用三参数reducing工厂方法实现的。它把流中的每个元素都转换成一个值为1的Long型对象，然后再把它们相加：

public static <T> Collector<T, ?, Long> counting() {
    return reducing(0L, e -> 1L, Long::sum);
}
1
2
3

使用泛型?通配符

?通配符，它用作counting工厂方法返回的收集器签名中的第二个泛型类型
经常使用Java的集合框架的话。在这里，它仅仅意味着收集器的累加器类型未知，换句话说，累加器本身可以是任何类型。我们在这里原封不动地写出了Collectors类中原始定义的方法签名

int totalCalories = menu.stream().map(Dish::getCalories).reduce(Integer::sum).get();




1
2
3
4
5

流的任何单参数reduce操作一样，reduce(Integer::sum)返回的不是int而是Optional，
以便在空流的情况下安全地执行归约操作。然后你只需用Optional对象中的get方法来提取里面的值就行了。
请注意，在这种情况下使用get方法是安全的，只是已经确定菜肴流不为空。
一般来说，使用允许提供默认值的方法，如orElse或orElseGet来解开Optional中包含的值更为安全
最后，更简洁的方法是把流映射到一个IntStream，然后调用sum方法，你也可以得到相同的结果：

int totalCalories = menu.stream().mapToInt(Dish::getCalories).sum();
1

计算菜单的总热量，使用IntStream，因为它最简明，也很可能最易读。同时，它也是性能最好的一个，因为IntStream可以让我们避免自动拆箱操作，也就是从Integer到int的隐式转换，它在这里毫无用处。
所以，根据情况选择最佳解决方案

String shortMenu = menu.stream().map(Dish::getName).collect(joining());
//替代1
String shortMenu = menu.stream().map(Dish::getName)
                 .collect( reducing     ( (s1, s2) -> s1 + s2 ) ).get();
//替代2
String shortMenu = menu.stream()
.collect( reducing( "",Dish::getName, (s1, s2) -> s1 + s2 ) );
1
2
3
4
5
6
7

分组

Map<Dish.Type, List<Dish>> dishesByType =
                      menu.stream().collect(groupingBy(Dish::getType));
//结果
{FISH=[prawns, salmon], OTHER=[french fries, rice, season fruit, pizza],
 MEAT=[pork, beef, chicken]}
//给groupingBy方法传递了一个Function（以方法引用的形式），它提取了流中每一道Dish的Dish.Type。
//我们把这个Function叫作分类函数，因为它用来把流中的元素分成不同的组
1
2
3
4
5
6
7

分类函数不一定像方法引用那样可用，因为你想用以分类的条件可能比简单的属性访问器要复杂。
例如，你可能想把热量不到400卡路里的菜划为“低热量”（diet），把热量在400到700卡路里之间的菜划为“普通”（normal），而把高于700卡路里的菜划为“高热量”（fat）。由于Dish类的没有把这个操作写成一个方法，因此无法使用方法引用，但可以把这个逻辑写成Lambda表达式

public enum CaloricLevel { DIET, NORMAL, FAT }
Map<CaloricLevel, List<Dish>> dishesByCaloricLevel = menu.stream().collect(
        groupingBy(dish -> {
                if (dish.getCalories() <= 400) return CaloricLevel.DIET;
                else if (dish.getCalories() <= 700) return CaloricLevel.NORMAL;
                else return CaloricLevel.FAT;
        } ));
1
2
3
4
5
6
7

操作分组的元素

按照菜肴的热量进行过滤操作，譬如找出那些热量大于500卡路里的菜肴
可以只加一个谓词进行过滤
Map<Dish.Type, List<Dish>> caloricDishesByType =
                    menu.stream().filter(dish -> dish.getCalories() > 500)
                                 .collect(groupingBy(Dish::getType));
//缺陷。如果你试着用它处理菜单，得到的结果是下面这种Map
{OTHER=[french fries, pizza], MEAT=[pork, beef]}
//没有任何一道类型是FISH的菜符合我们的过滤谓词，这个键在结果映射中完全消失了
1
2
3
4
5
6
7
8

Collectors类重载了工厂方法groupingBy。
除了常见的分类函数，它的第二变量也接受一个Collector类型的参数。通过这种方式，我们把过滤谓词挪到了第二个Collector中。

Map<Dish.Type, List<Dish>> caloricDishesByType =
        menu.stream()
            .collect(groupingBy(Dish::getType,
                     filtering(dish -> dish.getCalories() > 500, toList())));

filtering方法也是Collectors类的一个静态工厂方法，
它接受一个谓词对每一个分组中的元素执行过滤操作，
可以更进一步地使用Collector对过滤的元素继续进行分组。
通过这种方式，结果映射中依旧保存了FISH类型的条目，即便它映射的是一个空的列表：
//结果
{OTHER=[french fries, pizza], MEAT=[pork, beef], FISH=[]}
1
2
3
4
5
6
7
8
9
10
11

操作分组元素的另一种常见做法是使用一个映射函数对它们进行转换，这种方式也很有效。为了达成这个目标，Collectors类通过mapping方法提供了另一个Collector函数，它接受一个映射函数和另一个Collector函数作为参数。作为参数的Collector会收集对每个元素执行该映射函数的运行结果。这与之前看到的过滤收集器很相似。使用新的方法，可以将每道菜肴的分类添加到它们各自的菜名中。

Map<Dish.Type, List<String>> dishNamesByType = menu.stream()
					.collect(groupingBy(Dish::getType,mapping(Dish::getName, toList())));
1
2

这个例子中，结果映射的每个分组是一个由字符串构成的列表，而不是前面示例中的Dish类型。你还可以使用第三个Collector搭配groupingBy，再进行一次flatMap转换，这样得到的就不是一个普通的映射了。
为了演示这种机制是如何工作的，假设我们有一个映射，它为每道菜肴关联了一个标签列表

Map<String, List<String>> dishTags = new HashMap<>();
dishTags.put("pork", asList("greasy", "salty"));
dishTags.put("beef", asList("salty", "roasted"));
dishTags.put("chicken", asList("fried", "crisp"));
dishTags.put("french fries", asList("greasy", "fried"));
dishTags.put("rice", asList("light", "natural"));
dishTags.put("season fruit", asList("fresh", "natural"));
dishTags.put("pizza", asList("tasty", "salty"));
dishTags.put("prawns", asList("tasty", "roasted"));
dishTags.put("salmon", asList("delicious", "fresh"));
1
2
3
4
5
6
7
8
9
10

需要提取出每组菜肴对应的标签，使用flatMapping Collector可以轻松实现

Map<Dish.Type, Set<String>> dishNamesByType =
    menu.stream()
        .collect(groupingBy(Dish::getType,
                 flatMapping(dish -> dishTags.get( dish.getName() ).stream(),
                             toSet())));
//执行一个flatMap操作，将两层的结果列表归并为一层。
//注意，这一次我们会将每一组flatMapping操作的结果保存到一个Set中，而不是之前的List中，
//这么做是为了避免同一类型的多道菜由于关联了同样的标签而导致标签重复出现在结果集中
{MEAT=[salty, greasy, roasted, fried, crisp], FISH=[roasted, tasty, fresh,
 delicious], OTHER=[salty, greasy, natural, light, tasty, fresh, fried]}
1
2
3
4
5
6
7
8
9
10

多级分组

要实现多级分组，可以使用一个由双参数版本的Collectors.groupingBy工厂方法创建的收集器，它除了普通的分类函数之外，还可以接受collector类型的第二个参数。那么要进行二级分组的话，可以把一个内层groupingBy传递给外层groupingBy，并定义一个为流中项目分类的二级标准。

Map<Dish.Type, Map<CaloricLevel, List<Dish>>> dishesByTypeCaloricLevel = menu.stream().collect(
      groupingBy(Dish::getType,       ←---- 一级分类函数
          groupingBy(dish -> {       ←---- 二级分类函数
                  if (dish.getCalories() <= 400) return CaloricLevel.DIET;
                  else if (dish.getCalories() <= 700) return CaloricLevel.NORMAL;
                  else return CaloricLevel.FAT;
          } )
      )
);

//这个二级分组的结果就是像下面这样的两级Map：

{MEAT={DIET=[chicken], NORMAL=[beef], FAT=[pork]},
 FISH={DIET=[prawns], NORMAL=[salmon]},
 OTHER={DIET=[rice, seasonal fruit], NORMAL=[french fries, pizza]}}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

这里的外层Map的键就是第一级分类函数生成的值：“fish, meat, other”，而这个Map的值又是一个Map，键是二级分类函数生成的值：“normal, diet, fat”。
最后，第二级Map的值是流中元素构成的List，是分别应用第一级和第二级分类函数所得到的对应第一级和第二级键的值：“salmon，pizza…” 这种多级分组操作可以扩展至任意层级，n级分组就会得到一个代表n级树形结构的n级Map

n层嵌套映射和n维分类表之间的等价关系

把groupingBy看作“桶”比较容易明白。第一个groupingBy给每个键建立了一个桶。然后再用下游的收集器去收集每个桶中的元素，以此得到n级分组。
进一步说，传递给第一个groupingBy的第二个收集器可以是任何类型，而不一定是另一个groupingBy

Map<Dish.Type, Long> typesCount = menu.stream().collect(
                    groupingBy(Dish::getType, counting()));
//{MEAT=3, FISH=2, OTHER=4}
1
2
3

注意，普通的单参数groupingBy(f)（其中f是分类函数）实际上是groupingBy(f, toList())的简便写法。

Map<Dish.Type, Optional<Dish>> mostCaloricByType =
    menu.stream()
        .collect(groupingBy(Dish::getType,
                            maxBy(comparingInt(Dish::getCalories))));
1
2
3
4

注意　
这个Map中的值是Optional，因为这是maxBy工厂方法生成的收集器的类型，但实际上，如果菜单中没有某一类型的Dish，这个类型就不会对应一个Optional.empty()值，而且根本不会出现在Map的键中。groupingBy收集器只有在应用分组条件后，第一次在流中找到某个键对应的元素时才会把键加入分组Map中。这意味着Optional包装器在这里不是很有用，因为它不会仅仅因为是归约收集器的返回类型而表达一个最终可能不存在却意外存在的值。
分组操作的Map结果中的每个值上包装的Optional没什么用，所以你可能想要把它们去掉。要做到这一点，或者更一般地来说，把收集器返回的结果转换为另一种类型，你可以使用Collectors.collectingAndThen工厂方法返回的收集器。

Map<Dish.Type, Dish> mostCaloricByType =
    menu.stream()
        .collect(groupingBy(Dish::getType,       ←---- 分类函数
                 collectingAndThen(
                   maxBy(comparingInt(Dish::getCalories)),       ←---- 包装后的收集器
                 Optional::get)));       ←---- 转换函数
//结果
{FISH=salmon, OTHER=pizza, MEAT=pork}
1
2
3
4
5
6
7
8

这个工厂方法接受两个参数——要转换的收集器以及转换函数，并返回另一个收集器。这个收集器相当于旧收集器的一个包装，collect操作的最后一步就是将返回值用转换函数做一个映射。在这里，被包起来的收集器就是用maxBy建立的那个，而转换函数Optional::get则把返回的Optional中的值提取出来。
这个操作放在这里是安全的，因为reducing收集器永远都不会返回Optional.empty()。

把好几个收集器嵌套起来很常见，它们之间发生了什么可能不那么明显

• 收集器用虚线表示，因此groupingBy是最外层，根据菜肴的类型把菜单流分组，得到三个子流。
• groupingBy收集器包裹着collectingAndThen收集器，因此分组操作得到的每个子流都用这第二个收集器做进一步归约。
• collectingAndThen收集器又包裹着第三个收集器maxBy。
• 随后由归约收集器进行子流的归约操作，然后包含它的collectingAndThen收集器会对其结果应用Optional:get转换函数。
• 对三个子流分别执行这一过程并转换而得到的三个值，也就是各个类型中热量最高的Dish，将成为groupingBy收集器返回的Map中与各个分类键（Dish的类型）相关联的值。

Map<Dish.Type, Integer> totalCaloriesByType =
               menu.stream().collect(groupingBy(Dish::getType,
                       summingInt(Dish::getCalories)));
1
2
3

groupingBy联合使用的另一个收集器是mapping方法生成的。这个方法接受两个参数：一个函数对流中的元素做变换，另一个则将变换的结果对象收集起来。其目的是在累加之前对每个输入元素应用一个映射函数，这样就可以让接受特定类型元素的收集器适应不同类型的对象。

Map<Dish.Type, Set<CaloricLevel>> caloricLevelsByType =
menu.stream().collect(
    groupingBy(Dish::getType, mapping(dish -> {
            if (dish.getCalories() <= 400) return CaloricLevel.DIET;
            else if (dish.getCalories() <= 700) return CaloricLevel.NORMAL;
            else return CaloricLevel.FAT; },
    toSet() )));
//传递给映射方法的转换函数将Dish映射成了它的CaloricLevel：
//生成的CaloricLevel流传递给一个toSet收集器，它和toList类似，
//不过是把流中的元素累积到一个Set而不是List中，以便仅保留各不相同的值
//结果
{OTHER=[DIET, NORMAL], MEAT=[DIET, NORMAL, FAT], FISH=[DIET, NORMAL]}
1
2
3
4
5
6
7
8
9
10
11
12

对于返回的Set是什么类型并没有任何保证。
但通过使用toCollection，你就可以有更多的控制。你可以给它传递一个构造函数引用来要求HashSet。

Map<Dish.Type, Set<CaloricLevel>> caloricLevelsByType =
menu.stream().collect(
    groupingBy(Dish::getType, mapping(dish -> {
            if (dish.getCalories() <= 400) return CaloricLevel.DIET;
            else if (dish.getCalories() <= 700) return CaloricLevel.NORMAL;
            else return CaloricLevel.FAT; },
    toCollection(HashSet::new) )));
1
2
3
4
5
6
7

分区

分区是分组的特殊情况：由一个谓词（返回一个布尔值的函数）作为分类函数，它称分区函数。分区函数返回一个布尔值，这意味着得到的分组Map的键类型是Boolean，于是它最多可以分为两组——true是一组，false是一组。
分区的好处在于保留了分区函数返回true或false的两套流元素列表

Map<Boolean, List<Dish>> partitionedMenu =
             menu.stream().collect(partitioningBy(Dish::isVegetarian));
←---- 分区函数
//结果
{false=[pork, beef, chicken, prawns, salmon],
 true=[french fries, rice, season fruit, pizza]}
//找出所有的素食菜肴
List<Dish> vegetarianDishes = partitionedMenu.get(true);

注意，用同样的分区谓词，对菜单List创建的流作筛选，
然后把结果收集到另外一个List中也可以获得相同的结果：

List<Dish> vegetarianDishes =
            menu.stream().filter(Dish::isVegetarian).collect(toList());

可以使用两个筛选操作来访问partitionedMenu这个Map中false键的值：
一个利用谓词，一个利用该谓词的非。而且就像你在分组中看到的，
partitioningBy工厂方法有一个重载版本，可以像下面这样传递第二个收集器：
Map<Boolean, Map<Dish.Type, List<Dish>>> vegetarianDishesByType =
menu.stream().collect(
        partitioningBy(Dish::isVegetarian,       ←---- 分区函数
                       groupingBy(Dish::getType)));       ←---- 第二个收集器
结果
{false={FISH=[prawns, salmon], MEAT=[pork, beef, chicken]},
 true={OTHER=[french fries, rice, season fruit, pizza]}}

找到素食和非素食中热量最高的菜：
Map<Boolean, Dish> mostCaloricPartitionedByVegetarian =
menu.stream().collect(
    partitioningBy(Dish::isVegetarian,
        collectingAndThen(maxBy(comparingInt(Dish::getCalories)),
                          Optional::get)));
这将产生以下结果：
{false=pork, true=pizza}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

groupingBy收集器类似，partitioningBy收集器也可以结合其他收集器使用。尤其是它可以与第二个partitioningBy收集器一起使用来实现多级分区。

(1)
menu.stream().collect(partitioningBy(Dish::isVegetarian,
                      partitioningBy(d -> d.getCalories() > 500)));
(2)
menu.stream().collect(partitioningBy(Dish::isVegetarian,
                      partitioningBy(Dish:: getType)));
(3)
menu.stream().collect(partitioningBy(Dish::isVegetarian,
                      counting()));
答案：

(1) 这是一个有效的多级分区，产生以下二级Map：
{false={false=[chicken, prawns, salmon], true=[pork, beef]},
true={false=[rice, season fruit], true=[french fries, pizza]}}

(2) 这无法编译，因为partitioningBy需要一个谓词，
也就是返回一个布尔值的函数。方法引用Dish::getType不能用作谓词。

(3) 它会计算每个分区中项目的数目，得到以下Map：
{false=5, true=4}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

将数字按质数和非质数分区

写一个方法，它接受参数n（int类型），并将前n个自然数分为质数和非质数。
但首先，找出能够测试某一个待测数字是否是质数的谓词会很有帮助：

public boolean isPrime(int candidate) {
    return IntStream.range(2, candidate)       ←---- 产生一个自然数范围，从2开始，直至但不包括待测数
                    .noneMatch(i -> candidate % i == 0);       ←---- 如果待测数字不能被流中任何数字整除则返回true
}
一个简单的优化是仅测试小于等于待测数平方根的因子：

public boolean isPrime(int candidate) {
    int candidateRoot = (int) Math.sqrt((double) candidate);
    return IntStream.rangeClosed(2, candidateRoot)
                    .noneMatch(i -> candidate % i == 0);
}
现在最主要的一部分工作已经做好了。为了把前n个数字分为质数和非质数，
只要创建一个包含这n个数的流，用刚刚写的isPrime方法作为谓词，
再给partitioningBy收集器归约就好了：

public Map<Boolean, List<Integer>> partitionPrimes(int n) {
    return IntStream.rangeClosed(2, n).boxed()
                    .collect(
                        partitioningBy(candidate -> isPrime(candidate)));
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

Collectors类的静态工厂方法

收集器接口

public interface Collector<T, A, R> {
    Supplier<A> supplier();
    BiConsumer<A, T> accumulator();
    Function<A, R> finisher();
    BinaryOperator<A> combiner();
    Set<Characteristics> characteristics();
}
1
2
3
4
5
6
7

本列表适用以下定义。

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

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

public class ToListCollector<T> implements Collector<T, List<T>, List<T>>
1
2
3

四个方法都会返回一个会被collect方法调用的函数，第五个方法characteristics则提供了一系列特征，也就是一个提示列表，告诉collect方法在执行归约操作的时候可以应用哪些优化（比如并行化）。

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

supplier方法必须返回一个结果为空的Supplier，也就是一个无参数函数，在调用时它会创建一个空的累加器实例，供数据收集过程使用。很明显，对于将累加器本身作为结果返回的收集器，比如我们的ToListCollector，在对空流执行操作的时候，这个空的累加器也代表了收集过程的结果

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

将元素添加到结果容器：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;
}
1
2
3
4
5
6
7
8
9

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

在遍历完流后，finisher方法必须返回在累积过程的最后要调用的一个函数，以便将累加器对象转换为整个集合操作的最终结果。通常，就像ToListCollector的情况一样，累加器对象恰好符合预期的最终结果，因此无须进行转换。所以finisher方法只需返回identity函数：

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

这三个方法已经足以对流进行顺序归约，至少从逻辑上看可以下图进行。实践中的实现细节可能还要复杂一点，一方面是因为流的延迟性质，可能在collect操作之前还需要完成其他中间操作的流水线，另一方面则是理论上可能要进行并行归约。(顺序归约过程的逻辑步骤)

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

combiner方法会返回一个供归约操作使用的函数，它定义了对流的各个子部分进行并行处理时，各个子部分归约所得的累加器要如何合并。对于toList而言，这个方法的实现非常简单，只要把从流的第二子部分收集到的项目列表加到遍历第一子部分时得到的列表后面就行了：

public BinaryOperator<List<T>> combiner() {
    return (list1, list2) -> {
        list1.addAll(list2);
        return list1; }
}
此方法，就可以对流进行并行归约了。它会用到Java 7中引入的分支/合并框架和Spliterator抽象
1
2
3
4
5
6

小结

原始流会以递归方式拆分为子流，直到定义流是否需要进一步拆分的一个条件为非
（如果分布式工作单位太小，并行计算往往比顺序计算要慢，而且要是生成的并行任务比处理器内核数多很多的话就毫无意义了）。
现在，所有的子流都可以并行处理，即对每个子流应用图的顺序归约算法。最后，使用收集器combiner方法返回的函数，将所有的部分结果两两合并。这时会把原始流每次拆分时得到的子流对应的结果合并起来。(下图：顺序归约算法)
使用combiner方法来并行化归约过程

characteristics方法

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

• UNORDERED——归约结果不受流中项目的遍历和累积顺序的影响。
• CONCURRENT——accumulator函数可以从多个线程同时调用，且该收集器可以并行归约流。如果收集器没有标为UNORDERED，那它仅在用于无序数据源时才可以并行归约。
• IDENTITY_FINISH——这表明完成器方法返回的函数是一个恒等函数，可以跳过。这种情况下，累加器对象将会直接用作归约过程的最终结果。这也意味着，将累加器A不加检查地转换为结果R是安全的。

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

import java.util.*;
import java.util.function.*;
import java.util.stream.Collector;
import static java.util.stream.Collector.Characteristics.*;
public class ToListCollector<T> implements Collector<T, List<T>, List<T>> {
    @Override
    public Supplier<List<T>> supplier() {
        return ArrayList::new;       ←---- 创建集合操作的起始点
    }
    @Override
    public BiConsumer<List<T>, T> accumulator() {
        return List::add;       ←---- 累积遍历过的项目，原位修改累加器
    }
    @Override
    public Function<List<T>, List<T>> finisher() {
        return Function.identity();       ←---- 恒等函数
    }
    @Override
    public BinaryOperator<List<T>> combiner() {
        return (list1, list2) -> {
            list1.addAll(list2);       ←---- 修改第一个累加器，将其与第二个累加器的内容合并
            return list1;       ←---- 返回修改后的第一个累加器
        };
    }
    @Override
    public Set<Characteristics> characteristics() {
        return Collections.unmodifiableSet(EnumSet.of(
            IDENTITY_FINISH, CONCURRENT));       ←---- 为收集器添加IDENTITY_FINISH和CONCURRENT标志
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

注意，这个实现与Collectors.toList方法并不完全相同，但区别仅仅是一些小的优化。
这些优化的一个主要方面是Java API所提供的收集器在需要返回空列表时,
使用了Collections.emptyList()这个单例（singleton）。
这意味着它可安全地替代原生Java，来收集菜单流中的所有Dish的列表：

List<Dish> dishes = menuStream.collect(new ToListCollector<Dish>());
这个实现和标准的

List<Dish> dishes = menuStream.collect(toList());
构造之间的其他差异在于，toList是一个工厂，
而ToListCollector必须用new来实例化进行自定义收集而不去实现Collector

对于IDENTITY_FINISH的收集操作，还有一种方法可以得到同样的结果而无须从头实现新的Collector接口。
Stream有一个重载的collect方法可以接受另外三个函数——supplier、accumulator和combiner，
其语义和Collector接口的相应方法返回的函数完全相同。
List<Dish> dishes = menuStream.collect(
                        ArrayList::new,       ←---- 供应源
                        List::add,       ←---- 累加器
                        List::addAll);       ←---- 组合器
第二种形式虽然比前一个写法更为紧凑和简洁，却不那么易读。
此外，以恰当的类来实现自己的自定义收集器有助于重用并可避免代码重复。
另外值得注意的是，这第二个collect方法不能传递任何Characteristics，
所以它永远都是一个IDENTITY_FINISH和CONCURRENT但并非UNORDERED的收集器。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

自定义收集器，更好的性能

假设你有这个列表，那就可以把它传给isPrime方法，将方法重写如下：

public static boolean isPrime(List<Integer> primes, int candidate) {
    return primes.stream().noneMatch(i -> candidate % i == 0);
}
而且还应该应用先前的优化，仅仅用小于被测数平方根的质数来测试。
因此，你需要想办法在下一个质数大于被测数平方根时立即停止测试。
可以使用Stream的takeWhile的方法：

public static boolean isPrime(List<Integer> primes, int candidate){
    int candidateRoot = (int) Math.sqrt((double) candidate);
    return primes.stream()
                 .takeWhile(i -> i <= candidateRoot)
                 .noneMatch(i -> candidate % i == 0);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

以上是Java9之后的

实现自己的takeWhile方法，它接受一个排序列表和一个谓词，
返回列表元素中符合该谓词条件的最长子列表，代码如下所示：

public static <A> List<A> takeWhile(List<A> list, Predicate<A> p) {
    int i = 0;
    for (A item : list) {
            if (!p.test(item)) {       ←---- 检查列表中的当前元素是否符合谓词的约束
                    return list.subList(0, i);       ←---- 如果当前元素不符合谓词要求，返回测试元素的前序子列表
            }
            i++;
    }
    return list;       ←---- 列表中的所有元素都符合该谓词时，返回该列表
}

采用这种方式，你可以重写isPrime方法，只对那些不大于其平方根的候选素数进行测试：

public static boolean isPrime(List<Integer> primes, int candidate){
    int candidateRoot = (int) Math.sqrt((double) candidate);
    return takeWhile(primes, i -> i <= candidateRoot)
                .stream()
                .noneMatch(p -> candidate % p == 0);
}

        注意，与Stream API提供的版本不同，采用这种方式实现的版本是即时的。
        理想情况下，我们更希望采用Java 9那种由Stream提供的takeWhile，
        它具有延迟求值的特性，还能结合noneMatch来操作
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

声明一个实现Collector接口的新类

实现Collector接口所需的五个方法

Collector接口的定义是：

public interface Collector<T, A, R>
其中T、A和R分别是流中元素的类型、用于累积部分结果的对象类型，
以及collect操作最终结果的类型。这里应该收集Integer流，
而累加器和结果类型则都是Map<Boolean, List<Integer>>，键是true和false，
值则分别是质数和非质数的List:

public class PrimeNumbersCollector
             implements Collector<Integer,       ←---- 流中元素的类型
                                 Map<Boolean, List<Integer>>,       ←---- 累加器类型
                                 Map<Boolean, List<Integer>>>       ←---- collect操作的结果类型
1
2
3
4
5
6
7
8
9
10
11
12

接下来，你需要实现Collector接口中声明的五个方法。

supplier方法会返回一个在调用时创建累加器的函数：

public Supplier<Map<Boolean, List<Integer>>> supplier() {
    return () -> new HashMap<Boolean, List<Integer>>() {{
        put(true, new ArrayList<Integer>());
        put(false, new ArrayList<Integer>());
    }};
}
这里不但创建了用作累加器的Map，还为true和false两个键初始化了对应的空列表。
在收集过程中会把质数和非质数分别添加到这里。
收集器中最重要的方法是accumulator，因为它定义了如何收集流中元素的逻辑。
这里它也是实现前面所讲的优化的关键。
现在在任何一次迭代中，都可以访问收集过程的部分结果，也就是包含迄今找到的质数的累加器：

public BiConsumer<Map<Boolean, List<Integer>>, Integer> accumulator() {
    return (Map<Boolean, List<Integer>> acc, Integer candidate) -> {
            acc.get( isPrime(acc.get(true), candidate) )       ←---- 根据isPrime的结果，获取质数或非质数列表
               .add(candidate);       ←---- 将被测数添加到相应的列表中
    };
}
在这个方法中，调用了isPrime方法，将待测试是否为质数的数以及迄今找到的质数列表
（也就是累积Map中true键对应的值）传递给它。
这次调用的结果随后被用作获取质数或非质数列表的键，这样就可以把新的被测数添加到恰当的列表中
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

下一个方法要在并行收集时把两个部分累加器合并起来，
这里，它只需要合并两个Map，
即将第二个Map中质数和非质数列表中的所有数字合并到第一个Map的对应列表中就行了：

public BinaryOperator<Map<Boolean, List<Integer>>> combiner() {
    return (Map<Boolean, List<Integer>> map1,
            Map<Boolean, List<Integer>> map2) -> {
                map1.get(true).addAll(map2.get(true));
                map1.get(false).addAll(map2.get(false));
                return map1;
    };
}
请注意，实际上这个收集器是不能并行使用的，因为该算法本身是顺序的。
这意味着永远都不会调用combiner方法，
可以把它的实现留空（更好的做法是抛出一个UnsupportedOperationException异常）。
为了让这个例子完整，还是决定实现它。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

最后两个方法的实现都很简单。accumulator正好就是收集器的结果，
用不着进一步转换，那么finisher方法就返回identity函数：

public Function<Map<Boolean, List<Integer>>,
                Map<Boolean, List<Integer>>> finisher() {
        return Function.identity();
}
就characteristics方法而言，它既不是CONCURRENT也不是UNORDERED，却是IDENTITY_FINISH的：

public Set<Characteristics> characteristics() {
    return Collections.unmodifiableSet(EnumSet.of(IDENTITY_FINISH));
}
1
2
3
4
5
6
7
8
9
10
11
12

public class PrimeNumbersCollector
    implements Collector<Integer,
            Map<Boolean, List<Integer>>,
            Map<Boolean, List<Integer>>> {
    @Override
    public Supplier<Map<Boolean, List<Integer>>> supplier() {
        return () -> new HashMap<Boolean, List<Integer>>() {{       ←---- 从一个有两个空List的Map开始收集过程
            put(true, new ArrayList<Integer>());
            put(false, new ArrayList<Integer>());
          }};
    }
    @Override
    public BiConsumer<Map<Boolean, List<Integer>>, Integer> accumulator() {
        return (Map<Boolean, List<Integer>> acc, Integer candidate) -> {
            acc.get( isPrime( acc.get(true),       ←---- 将已经找到的质数列表传递给isPrime方法
               candidate) )
               .add(candidate);       ←---- 根据isPrime方法的返回值，从Map中取质数或非质数列表，把当前的被测数加进去
        };
    }
    @Override
    public BinaryOperator<Map<Boolean, List<Integer>>> combiner() {
            return (Map<Boolean, List<Integer>> map1,
                    Map<Boolean, List<Integer>> map2) -> {       ←---- 将第二个Map合并到第一个
                            map1.get(true).addAll(map2.get(true));
                            map1.get(false).addAll(map2.get(false));
                            return map1;
                };
    }
    @Override
    public Function<Map<Boolean, List<Integer>>,
                    Map<Boolean, List<Integer>>> finisher() {
                            return Function.identity();       ←---- 收集过程最后无须转换，因此用identity函数收尾
    }
    @Override
    public Set<Characteristics> characteristics() {
        return Collections.unmodifiableSet(EnumSet.of(IDENTITY_FINISH));       ←---- 这个收集器是IDENTITY_FINISH，但既不是UNORDERED也不是CONCURRENT，因为质数是按顺序发现的
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

比较器性能

public class CollectorHarness {
    public static void main(String[] args) {
        long fastest = Long.MAX_VALUE;
        for (int i = 0; i < 10; i++) {
            long start = System.nanoTime();       ←---- 运行测试10次
            partitionPrimes(1_000_000);       ←---- 将前一百万个自然数按质数和非质数分区
            long duration = (System.nanoTime() - start) / 1_000_000;       ←---- 取运行时间的毫秒值
            if (duration < fastest) fastest = duration;       ←---- 检查这个执行是否是最快的一个
        }
        System.out.println(
            "Fastest execution done in " + fastest + " msecs");
    }
}
    这个类会先把前一百万个自然数分为质数和非质数，
    利用partitioningBy工厂方法创建的收集器调用方法10次，记下最快的一次运行
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

public Map<Boolean, List<Integer>> partitionPrimesWithCustomCollector
        (int n) {
    IntStream.rangeClosed(2, n).boxed()
        .collect(
            () -> new HashMap<Boolean, List<Integer>>() {{       ←---- 供应源
                put(true, new ArrayList<Integer>());
                put(false, new ArrayList<Integer>());
            }},
            (acc, candidate) -> {       ←---- 累加器
                acc.get( isPrime(acc.get(true), candidate) )
                   .add(candidate);
            },
            (map1, map2) -> {       ←---- 组合器
                map1.get(true).addAll(map2.get(true));
                map1.get(false).addAll(map2.get(false));
            });
}
这样就可以避免为实现Collector接口创建一个全新的类；
得到的代码更紧凑，虽然可能可读性会差一点，可重用性会差一点。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

小结

• collect是一个终端操作，它接受的参数是将流中元素累积到汇总结果的各种方式（称为收集器）。
• 预定义收集器包括将流元素归约和汇总到一个值，例如计算最小值、最大值或平均值。
• 预定义收集器可以用groupingBy对流中元素进行分组，或用partitioningBy进行分区。
• 收集器可以高效地复合起来，进行多级分组、分区和归约。
• 实现Collector接口中定义的方法来自定义的收集器。