在 Collectors.groupingBy 上应用自定义聚合

Question

我有一个

Map<A,B> AvsB

，其中

A

和

B

是，

class A {
    Long id;
    AggregationType aggr;
}

class B {
    Long value;
}

其中 AggregationType 是一个包含 (

SUM, AVG, MIN, MAX

) 的枚举。

我从条目集中创建了一个流，我想按 A.id 对这个条目列表进行分组，并对来自结果下游的 B.value 应用自定义聚合。

Map<Long, Double> aggregatedMap = AvsB.entrySet().stream().groupingBy(
    entry -> entry.getKey().getId(),
    Collectors.summingDouble(entry-> (double)entry.getValue().getValue())
)

我能够为单个聚合解决这个问题，比如本例中的 sum，但我不知道如何通过从 A 获取的聚合来解决它（即）我需要为每个枚举分配一个 switch case，所以对于例如，对于

AVG

，我将使用

Collectors.averagingDouble

而不是

Collectors.summingDouble

。 有人可以帮忙吗？谢谢。

Answer 1

有两个问题：首先，这些收集器没有相同的返回类型。即使你使用 DoubleStream 或 LongStream 你会更接近，但你仍然不会有完全相同的返回类型。因此，我们需要做一些额外的工作才能使它们返回相同的东西。例如，这可以是一个选项：

private double collectValues(AggregatorType agg, List<Long> values) {
    DoubleStream stream = values.stream().mapToDouble(x -> x + 0.0d);
    return switch (agg) {
        case AVG -> stream.average().orElseThrow();
        case SUM -> stream.sum();
        case MAX -> stream.max().orElseThrow();
        case MIN -> stream.min().orElseThrow();
        default -> throw new IllegalArgumentException();
    };
}

第二个问题是我们不能轻易地为同一流的元素使用不同的收集器。因此，您需要做一些不同的事情——例如，您可以分两步完成。最初，通过 Aggegator.Type 收集所有 B.value ：

public Map<AggregatorType, Double> aggregate(Map<A, B> fields) {
    Map<AggregatorType, List<Long>> valuesByType = fields.entrySet()
        .stream()
        .collect(Collectors.groupingBy(
            entry -> entry.getKey().type(),
            Collectors.mapping(
                entry -> entry.getValue().value(),
                Collectors.toList())
        ));
    // return valuesByType.stream()...
}    

然后，使用第一个代码片段中的函数收集每个 List：

 return valuesByType.entrySet()
     .stream()
     .collect(Collectors.toMap(
            entry -> entry.getKey(),
            entry -> collectValues(entry.getKey(), entry.getValue())
     ));

这里是完整的故事：

@Test
void test() {
    //given
    Map<A, B> values = Map.of(
        new A(1L, AggregatorType.AVG), new B(10L),
        new A(2L, AggregatorType.AVG), new B(20L),
        new A(3L, AggregatorType.SUM), new B(30L),
        new A(4L, AggregatorType.SUM), new B(40L),
        new A(5L, AggregatorType.MAX), new B(50L),
        new A(6L, AggregatorType.MAX), new B(60L),
        new A(7L, AggregatorType.MIN), new B(70L),
        new A(8L, AggregatorType.MIN), new B(80L)
    );

    // when
    Map<AggregatorType, Double> result = aggregate(values);

    //then
    assertThat(result).isEqualTo(Map.of(
        AggregatorType.AVG, 15d,
        AggregatorType.SUM, 70d,
        AggregatorType.MAX, 60d,
        AggregatorType.MIN, 70d
    ));
}

public Map<AggregatorType, Double> aggregate(Map<A, B> fields) {
    Map<AggregatorType, List<Long>> valuesByType = fields.entrySet()
        .stream()
        .collect(Collectors.groupingBy(
            entry -> entry.getKey().type(),
            Collectors.mapping(
                entry -> entry.getValue().value(),
                Collectors.toList())
        ));
    return valuesByType.entrySet()
        .stream()
        .collect(Collectors.toMap(
            entry -> entry.getKey(),
            entry -> collectValues(entry.getKey(), entry.getValue())
        ));
}

private double collectValues(AggregatorType aggregator, List<Long> values) {
    DoubleStream stream = values.stream().mapToDouble(x -> x + 0.0d);
    return switch (aggregator) {
        case AVG -> stream.average().orElseThrow();
        case SUM -> stream.sum();
        case MAX -> stream.max().orElseThrow();
        case MIN -> stream.min().orElseThrow();
        default -> throw new IllegalArgumentException();
    };
}

如果你想看看我是如何一步步做到的，请随时阅读这篇文章：https://medium.com/javarevisited/polymorphic-stream-collector-in-java-44f9008bf043