Java 8 中的嵌套收集器答案

【问题标题】：nesting collectors in Java 8Java 8 中的嵌套收集器
【发布时间】：2014-10-31 13:23:10
【问题描述】：

我正在处理人口统计数据。我有一个州不同县的记录集合（每个县有几条记录），我想按县汇总。

我已经实现了以下消费者：

public class CountyPopulation implements java.util.function.Consumer<Population>
{
    private String countyId ;
    private List<Demographic> demographics ;

    public CountyPopulation()
    {
        demographics = new ArrayList<Demographic>() ;
    }

    public List<Demographic> getDemographics()
    {
        return demographics ;
    }

    public void accept(Population pop) 
    {
        if ( countyId == null )
        {
            countyId = pop.getCtyId() ;
        }
        demographics.add( pop.getDemographic() ) ;
    }

    public void combine(CountyPopulation other) 
    {
        demographics.addAll( other.getDemographics() ) ;
    }
}

此 CountyPopulation 用于使用以下代码（其中“089”是县标识符）聚合有关特定县的数据：

CountyPopulation ctyPop = populations
    .stream()
    .filter( e -> "089".equals( e.getCtyId() ) )
    .collect(CountyPopulation::new, 
             CountyPopulation::accept, 
             CountyPopulation::combine) ;

现在，我想在使用我的聚合器之前删除“过滤器”并按县对记录进行分组。

根据您的第一个答案，我知道这可以通过以下方式使用静态函数 Collector.of 完成：

Map<String,CountyPopulation> pop = populations
    .stream()
    .collect(
        Collectors.groupingBy(Population::getCtyId,
                              Collector.of( CountyPopulation::new,
                                            CountyPopulation::accept, 
                                            (a,b)->{a.combine(b); return a;} ))) ;

但是，此代码不起作用，因为 Collector.of() 的签名与 collect() 不同。我怀疑该解决方案涉及修改类 CountyPopulation 以便它实现 java.util.function.BiConsumer 而不是 java.util.function.Consumer 但我这样做的尝试没有奏效，我不清楚为什么。

【问题讨论】：

标签： java mapreduce java-8

【解决方案1】：

在Stream 上调用collect with the three arguments 等效于使用Collector.of。

因此您可以使用以下方法实现您的目标：

Map<String,CountyPopulation> pop = populations.stream().collect(
  Collectors.groupingBy(Population::getCtyId, Collector.of(
    CountyPopulation::new, CountyPopulation::accept, CountyPopulation::combine))) ;

为了获得更好的并行性能，值得研究您可以提供的可选Characteristics。如果 UNORDERED 或 CONCURRENT 中的一个或两个与您的 CountyPopulation 类的行为匹配，您可以提供它们（IDENTITY_FINISH 在您的情况下是隐含的）。

使用groupingByConcurrent 代替groupingBy 也可以提高并行性能。

【讨论】：

我一定还是遗漏了一些东西，现在 Java 抱怨 of 方法的参数。 Collector 类型中的 (Supplier, BiConsumer, BinaryOperator, Collector.Characteristics...) 方法不适用于参数 (CountyPopulation::new, CountyPopulation::accept, CountyPopulation ::结合）
Collector.of 的签名与collect 略有不同。 collect 的第三个参数是 BiConsumer，但 Collector.of 需要 BiOperator。将CountyPopulation::combine 替换为(a,b)->{a.combine(b); return a;}。或者，或者，更改 CountyPopulation.combine 方法以返回结果 CountyPopulation。
@Misha：对，或者简单地说，如果语义没有改变，CountyPopulation.combine 必须返回this。

【解决方案2】：

好的，我终于让它工作了，但我必须显式添加特征参数：

ConcurrentMap<String,CountyPopulation> pop = populations
    .parallelStream().collect(
      Collectors.groupingByConcurrent(
          Population::getCtyId, 
          Collector.of( 
              CountyPopulation::new, 
              CountyPopulation::accept, 
              (a,b)-> {a.combine(b); return a; }, 
              Characteristics.IDENTITY_FINISH  ) ) )  ;

【讨论】：

如果你使用三参数方法，你不需要指定IDENTITY_FINISH，因为你没有指定完成函数，这是隐含的。您不需要指定过时的特征；这是一个 varargs 方法，它也不接受任何特征参数。