Java Stream 按字段分组并按该字段计数排序？

Question

我正在尝试通过在以下实体上使用 Java 流来获取拥有最大城市的 3 个国家中的

countryCode

：

城市：

id   | name     | countryCode |
------------------------------
1    | Berlin   | DE          |
2    | Munich   | DE          |
3    | Köln     | DE          |
4    | Paris    | FR          |
5    | Kopenhag | DK          |
...

我尝试了一些方法，但没有按预期工作。那么，获取前 3 个国家代码（前 3 个大多重复的国家代码）的最正确方法是什么？

final Map<String, City> airportCodes2 = airportService.getCities().stream()
                .map(City::getCountryCode, Function.identity())
                .toMap();

Answer 1

排序

最简单的方法是通过生成

countryCode

类型的辅助地图，将数据按

Map<String, Long>

进行分组，并将每个

countryCode

与城市数量相关联。

然后在地图条目上生成一个流，并按Value以相反的顺序对其进行排序（即按城市计数的降序排列）。

如何实施：

public static List<String> getTopNCodes(List<City> cities,
                                        int limit) {
    return cities.stream()
        .collect(Collectors.groupingBy( // creates a Map<String, Long>
            City::getCountryCode,
            Collectors.counting()
        ))
        .entrySet().stream()
        .sorted(Map.Entry.<String, Long>comparingByValue().reversed())
        .limit(limit) // <- retain top N
        .map(Map.Entry::getKey)
        .toList();
}

这种方法的时间复杂度是 O(n * log n) 不利于排序，如果检索它的元素数量很少（如

3

），而数据却很大，这将是不利的。

我们可以做得更好。

使用自定义收集器进行部分排序

我们可以使用

PriorityQueue

执行部分排序，而不是对辅助映射的所有条目进行排序（此类是 JDK 提供的 Binary Heap 的实现）。

为此，我们可以使用静态工厂方法实现自定义收集器

Collector.of()

。

PriorityQueue

的实例将用作收集器的可变容器。流中的每个映射条目都将与堆的 root 元素进行比较。如果元素更大（条目拥有更大的计数），则下一个条目将添加到队列根中。如果超过限制，则应删除根元素。

为了使代码可重用，我们可以对其进行绅士化。第一部分（使用代表键频率的值创建中间映射）保持不变。

public static <T, K> List<K> getTopN(List<T> list,
                                          Function<T, K> keyExtractor,
                                          int limit) {
    return list.stream()
        .collect(Collectors.groupingBy(
            keyExtractor,
            Collectors.counting()
        ))
        .entrySet().stream()
        .collect(getMaxN(
            limit,
            Map.Entry.<K, Long>comparingByValue().reversed(),
            Map.Entry::getKey
        ));
}

基于最小堆的收集器：

public static <T, R> Collector<T, ?, List<R>> getMaxN(int size,
                                                      Comparator<T> comparator,
                                                      Function<T, R> keyExtractor) {
    
    return Collector.of(
        () -> new PriorityQueue<>(comparator),
        (Queue<T> queue, T next) -> tryAdd(queue, next, comparator, size),
        (Queue<T> left, Queue<T> right) -> {
            right.forEach(next -> tryAdd(left, next, comparator, size));
            return left;
        },
        (Queue<T> queue) -> queue.stream().map(keyExtractor).toList(),
        Collector.Characteristics.UNORDERED
    );
}

public static <T> void tryAdd(Queue<T> queue, T next, Comparator<T> comparator, int size) {
    if (queue.size() == size && comparator.compare(queue.element(), next) < 0)
        queue.remove(); // if next value is greater than the smallest element in the queue and max size has been exceeded the smallest element needs to be removed from the queue
    if (queue.size() < size) queue.add(next);
}