如何利用Java Stream在数据库取数时实时执行分组计算？

你的问题戳中了一个很常见的痛点：当处理超大规模数据集时，先把所有数据加载到内存再分组统计，很容易导致内存溢出。原代码里的collect(Collectors.groupingBy(...))确实会先把所有1亿条数据都攒到内存里，然后再分组——这对于你的场景（Country+City基数低，但总数据量极大）来说完全没必要，我们可以改成每拿到一条数据就更新统计结果，根本不用存所有原始数据。

核心问题分析 #

原代码的流程是：

1. 并行执行10个查询，每个返回1000万条数据的Stream
2. flatMap把所有Stream合并成一个大Stream
3. groupingBy会先把所有元素收集到内存中的List，再按Country+City分组
4. 最后再遍历分组后的List计算总和

这一步的问题在第3步：所有1亿条数据都会被暂存，内存压力爆炸。我们需要把统计逻辑提前到收集阶段，而不是等所有数据都到位再处理。

解决方案：用Collectors.toMap实时累加统计 #

我们可以定义一个专门的统计类，用来存每个Country+City组合的总和，然后用Collectors.toMap在每一条数据流过来时就更新统计值，而不是保存原始数据。

第一步：定义统计类

先写一个类来封装每个分组的统计结果，包含累加和合并逻辑（并行流需要合并不同线程的统计结果）：

import java.util.Objects;

class CityCountryStats {
    private final String country;
    private final String city;
    private double sumField1 = 0.0;
    private double sumField2 = 0.0;
    private double sumField3 = 0.0;

    // 从单条数据初始化统计对象
    public CityCountryStats(String country, String city) {
        this.country = country;
        this.city = city;
    }

    // 累加单条数据的字段值
    public void accumulate(DataRow row) { // 这里的DataRow是你的数据库行对象类型
        this.sumField1 += row.getSumField1();
        this.sumField2 += row.getSumField2();
        this.sumField3 += row.getSumField3();
    }

    // 合并两个统计对象（并行流中不同线程的结果需要合并）
    public CityCountryStats merge(CityCountryStats other) {
        this.sumField1 += other.sumField1;
        this.sumField2 += other.sumField2;
        this.sumField3 += other.sumField3;
        return this;
    }

    // 转成你需要的MyResultClass
    public MyResultClass toMyResult() {
        MyResultClass result = new MyResultClass();
        result.setCountry(country);
        result.setCity(city);
        result.setSumField1(sumField1);
        result.setSumField2(sumField2);
        result.setSumField3(sumField3);
        return result;
    }
}

第二步：优化流处理逻辑

修改原有的流代码，用toMap替代groupingBy，实现实时统计：

import java.util.Arrays;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;
import java.util.stream.Collectors;

myqueries.parallelStream()
    .map(query -> queryresult) // 每个查询返回DataRow的Stream
    .flatMap(Function.identity())
    // 用toMap实现实时分组统计
    .collect(Collectors.toMap(
        // 分组键：这里可以用自定义类替代List，更可靠（后面会说）
        row -> Arrays.asList(row.getCountry(), row.getCity()),
        // 从单条数据创建统计对象
        row -> {
            CityCountryStats stats = new CityCountryStats(row.getCountry(), row.getCity());
            stats.accumulate(row);
            return stats;
        },
        // 同一个分组的统计对象合并
        CityCountryStats::merge,
        // 并行流必须用线程安全的Map，避免并发问题
        ConcurrentHashMap::new
    ))
    // 把统计对象转成你的MyResultClass并处理
    .values().stream()
    .map(CityCountryStats::toMyResult)
    .forEach(result -> {
        // 打印结果，示例：
        System.out.printf("Country: %s, City: %s, Sum1: %.2f, Sum2: %.2f, Sum3: %.2f%n",
            result.getCountry(), result.getCity(),
            result.getSumField1(), result.getSumField2(), result.getSumField3());
    });

额外优化：用自定义分组键替代List #

原代码用List<Object>作为分组键虽然可行，但用自定义的不可变类会更可靠，性能也更好（避免数组的hash计算）：

class CountryCityKey {
    private final String country;
    private final String city;

    public CountryCityKey(String country, String city) {
        this.country = country;
        this.city = city;
    }

    // 必须正确重写equals和hashCode
    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        CountryCityKey that = (CountryCityKey) o;
        return Objects.equals(country, that.country) && Objects.equals(city, that.city);
    }

    @Override
    public int hashCode() {
        return Objects.hash(country, city);
    }
}

然后把toMap的键部分改成：

row -> new CountryCityKey(row.getCountry(), row.getCity())

关键注意事项 #

1. 数据库查询必须是流式的：确保你的数据库驱动支持流式返回数据，比如JDBC中要设置Statement.setFetchSize(Integer.MIN_VALUE)，或者用ORM框架的Stream返回（比如Spring Data JPA的Stream<T>）。如果数据库驱动一次性把所有数据加载到客户端内存，那Java这边的流式处理也救不了内存溢出问题。
2. 并行流的线程安全：因为用了parallelStream，所以必须用ConcurrentHashMap作为toMap的容器，同时合并逻辑要保证线程安全（这里的merge方法是纯累加，没问题）。

这样修改后，内存占用只和Country+City的组合数有关，而不是1亿条原始数据，完美解决你的问题！

内容的提问来源于stack exchange，提问作者Fuat