Java 8中的嵌套收集器

Question

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

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

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.formation.Consumer，但我这样做的尝试没有奏效，我不清楚为什么。

Answer 1

在Stream上用三个参数调用collect等同于使用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在您的情况下是隐含的）。

使用CCD_ 9代替CCD_ 10也可以提高并行性能。

Answer 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  ) ) )  ;