优化Java数据批量获取：利用Stream API避免共享可变性

1. 问题背景与共享可变性挑战

在实际的软件开发中，我们经常需要从数据库中获取大量数据。然而，数据库通常对单个查询中允许的参数数量有限制（例如，sql in 子句可能限制在500个参数）。这意味着我们不能一次性查询所有数据，而需要将大的查询列表分割成多个小批次进行查询。

考虑以下场景：我们需要根据一个包含5000个数字的列表，分批从数据库中获取猫（Cat）和狗（Dog）的信息，每次查询最多接受500个参数。一个常见的初始实现方式可能如下：

// 假设 Cat 和 Dog 是实体类，catRepo 和 dogRepo 是数据访问层接口
// CatRepo 和 DogRepo 都有 fetchCats(List) 和 fetchDogs(List) 方法

AtomicInteger counter = new AtomicInteger();
List catList = new ArrayList<>(); // 共享的可变列表
List dogList = new ArrayList<>(); // 共享的可变列表

// 模拟生成5000个数字作为查询键
List numbers = Stream.iterate(1, e -> e + 1)
    .limit(5000)
    .collect(Collectors.toList());

// 将数字列表分割成每批次500个的小列表
Collection> partitionedListOfNumbers = numbers.stream()
    .collect(Collectors.groupingBy(num -> counter.getAndIncrement() / 500))
    .values(); // 得到 List> 结构

// 遍历分批的列表，执行查询并累加结果
partitionedListOfNumbers.stream()
    .forEach(list -> {
        List interimCatList = catRepo.fetchCats(list); // 从数据库获取猫列表
        catList.addAll(interimCatList); // 修改外部的 catList
        List interimDogList = dogRepo.fetchDogs(list); // 从数据库获取狗列表
        dogList.addAll(interimDogList); // 修改外部的 dogList
    });

// 此时 catList 和 dogList 包含了所有查询结果

上述代码虽然能够完成任务，但存在一个明显的问题：catList 和 dogList 是在 forEach 循环外部声明的，并且在循环内部通过 addAll 方法被修改。这种模式被称为共享可变性（Shared Mutability），即多个操作（或线程）共享并修改同一个可变状态。

共享可变性在函数式编程中被视为一种“副作用”，它带来了以下弊端：

• 难以推理：代码的行为不再仅仅取决于其输入，还取决于外部状态，使得理解和预测程序行为变得复杂。
• 并发安全问题：在多线程环境中，多个线程同时修改共享列表可能导致数据不一致或竞态条件，需要额外的同步机制（如 Collections.synchronizedList 或 CopyOnWriteArrayList），增加了复杂性。
• 可测试性差：由于依赖外部状态，单元测试变得困难，需要模拟或设置复杂的上下文。

为了提高代码的纯净性、可读性和并发安全性，我们应该尽量避免共享可变性。

立即学习“Java免费学习笔记（深入）”；

2. 基于Stream API的解决方案

Java 8 引入的 Stream API 提供了一种声明式、函数式的数据处理方式，非常适合解决上述问题，因为它鼓励通过转换（transformation）而不是修改（mutation）来处理数据。

核心思想是：将每个分批查询的结果视为一个独立的中间集合，然后将所有这些中间集合扁平化并收集到一个最终的不可变集合中。这可以通过 map、flatMap 和 collect 操作组合实现。

以下是重构后的代码示例：

Winston AI

强大的AI内容检测解决方案

下载

import java.util.Collection;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.function.Function; // 用于后续优化

// 假设 Cat, Dog, CatRepo, DogRepo 已经定义
// ... (Cat, Dog 实体类及 CatRepo, DogRepo 接口的定义)

public class BatchDataFetcher {

    // 假设 catRepo 和 dogRepo 已经通过依赖注入或其他方式初始化
    private CatRepo catRepo;
    private DogRepo dogRepo;

    public BatchDataFetcher(CatRepo catRepo, DogRepo dogRepo) {
        this.catRepo = catRepo;
        this.dogRepo = dogRepo;
    }

    public void fetchDataAndProcess() {
        // 用于分批的计数器，确保每个批次编号正确
        AtomicInteger counter = new AtomicInteger(); 

        // 模拟生成5000个数字，并将其分批
        // IntStream.rangeClosed(1, 5000) 生成从1到5000的整数流
        // .boxed() 将 IntStream 转换为 Stream
        Collection> partitionedListOfNumbers = IntStream.rangeClosed(1, 5000)
            .boxed() 
            .collect(Collectors.groupingBy(num -> counter.getAndIncrement() / 500))
            .values(); // 得到 List> 结构

        // 获取所有猫列表
        List catList = partitionedListOfNumbers.stream()
            .map(list -> catRepo.fetchCats(list)) // 对每个分批列表执行查询，得到 Stream>
            .flatMap(List::stream) // 将 Stream> 扁平化为 Stream
            .collect(Collectors.toList()); // 收集所有 Cat 对象到一个新的 List 中

        // 获取所有狗列表
        List dogList = partitionedListOfNumbers.stream()
            .map(list -> dogRepo.fetchDogs(list)) // 对每个分批列表执行查询，得到 Stream>
            .flatMap(List::stream) // 将 Stream> 扁平化为 Stream
            .collect(Collectors.toList()); // 收集所有 Dog 对象到一个新的 List 中

        // 此时 catList 和 dogList 是通过 Stream 操作“生成”的，而不是“修改”的
        // 它们是不可变的（如果 collect 收集到的是不可变列表，否则是新的可变列表，但不再是共享的）
        System.out.println("Fetched " + catList.size() + " cats and " + dogList.size() + " dogs.");
    }
}

// 模拟 Cat, Dog, CatRepo, DogRepo
class Cat { private int id; private String name; public Cat(int id, String name) { this.id = id; this.name = name; } @Override public String toString() { return "Cat{" + "id=" + id + ", name='" + name + '\'' + '}'; } }
class Dog { private int id; private String name; public Dog(int id, String name) { this.id = id; this.name = name; } @Override public String toString() { return "Dog{" + "id=" + id + ", name='" + name + '\'' + '}'; } }
class CatRepo { public List fetchCats(List ids) { return ids.stream().map(id -> new Cat(id, "Cat-" + id)).collect(Collectors.toList()); } }
class DogRepo { public List fetchDogs(List ids) { return ids.stream().map(id -> new Dog(id, "Dog-" + id)).collect(Collectors.toList()); } }

// 示例运行
// public static void main(String[] args) {
//     BatchDataFetcher fetcher = new BatchDataFetcher(new CatRepo(), new DogRepo());
//     fetcher.fetchDataAndProcess();
// }

代码解析：

1. 数据分批 (groupingBy)： IntStream.rangeClosed(1, 5000).boxed().collect(Collectors.groupingBy(num -> counter.getAndIncrement() / 500)).values() 这一步与原始代码类似，负责将连续的数字列表分割成多个子列表，每个子列表包含500个数字。counter 在这里作为分组键的一部分，确保每500个数字被分到同一个组。

2. map 操作： partitionedListOfNumbers.stream().map(list -> catRepo.fetchCats(list)) 这一步将 Collection> 转换成 Stream>。对于 partitionedListOfNumbers 中的每一个 List（即一个批次的查询键），catRepo.fetchCats(list) 会被调用，返回一个 List。map 操作的输出是一个包含多个 List 的流。

3. flatMap 操作： .flatMap(List::stream) 由于上一步 map 操作的输出是 Stream>，我们希望得到的是一个单一的 List。flatMap 的作用就是将流中的每个元素（这里是 List）“扁平化”成一个流（通过 List::stream 方法），然后将所有这些子流连接成一个单一的流。最终，我们得到的是一个 Stream。

4. collect 操作： .collect(Collectors.toList()) 这是流操作的终结操作，它将 Stream 中的所有 Cat 对象收集到一个新的 List 中。这个 List 是一个全新的列表，不与任何外部变量共享，从而避免了共享可变性。

3. 解决方案的优势

通过上述重构，我们获得了以下显著优势：

• 避免共享可变性：catList 和 dogList 不再在循环内部被修改，而是通过 Stream API 的一系列转换操作“生成”的最终结果。这使得代码更加纯净，更容易理解。
• 提升代码可读性与函数式风格：Stream API 提供了声明式编程风格，代码清晰地表达了“如何转换数据”而不是“如何一步步修改状态”，符合现代Java的函数式编程范式。
• 更好的并发安全性：由于没有共享的可变状态，此代码天然支持并发。如果将 partitionedListOfNumbers.stream() 替换为 partitionedListOfNumbers.parallelStream()，查询可以并行执行而无需担心竞态条件，因为每个 map 操作都是独立的，并且最终的 collect 操作会安全地将结果聚合。

4. 进一步优化：消除重复代码

在上述解决方案中，获取 catList 和 dogList 的逻辑非常相似，都遵循 map -> flatMap -> collect 的模式。为了进一步提高代码的复用性并消除重复，我们可以将这部分通用逻辑提取到一个泛型方法中，该方法接受一个函数作为参数，用于执行具体的数据库查询：

public class BatchDataFetcher {
    // ... (构造函数和成员变量不变)

    /**
     * 通用方法：根据分批的键列表和查询函数批量获取数据
     * @param partitionedKeys 分批的查询键列表
     * @param fetchFunction 接受一个 List 并返回 List 的查询函数
     * @param  结果列表中的元素类型
     * @param  查询键列表中的元素类型
     * @return 聚合后的所有 T 类型元素的列表
     */
    public  List fetchEntitiesInBatches(Collection> partitionedKeys, Function, List> fetchFunction) {
        return partitionedKeys.stream()
            .map(fetchFunction) // 对每个批次应用查询函数
            .flatMap(List::stream) // 扁平化结果
            .collect(Collectors.toList()); // 收集到新列表
    }

    public void fetchDataAndProcessOptimized() {
        AtomicInteger counter = new AtomicInteger();
        Collection> partitionedListOfNumbers = IntStream.rangeClosed(1, 5000)
            .boxed()
            .collect(Collectors.groupingBy(num -> counter.getAndIncrement() / 500))
            .values();

        // 使用通用方法获取猫列表
        List catList = fetchEntitiesInBatches(partitionedListOfNumbers, catRepo::fetchCats);

        // 使用通用方法获取狗列表
        List dogList = fetchEntitiesInBatches(partitionedListOfNumbers, dogRepo::fetchDogs);

        System.out.println("Optimized: Fetched " + catList.size() + " cats and " + dogList.size() + " dogs.");
    }
}

通过引入 fetchEntitiesInBatches 泛型方法，我们将批处理查询的核心逻辑抽象出来，使得 fetchDataAndProcessOptimized 方法更加简洁，并且易于扩展到其他类型的实体查询。

5. 总结与最佳实践

在Java中处理批量数据获取并避免共享可变性是一个常见的需求。通过本教程，我们学习到：

• 识别共享可变性问题：当代码通过 forEach 循环修改外部集合时，往往存在共享可变性问题，这会降低代码质量和并发安全性。
• 利用Stream API进行转换：Java Stream API 的 map、flatMap 和 collect 操作是解决这类问题的强大工具。它们允许我们以声明式的方式对数据进行一系列无副作用的转换，最终生成新的集合。
• 拥抱函数式编程：通过避免副作用和共享可变性，我们能够编写出更纯净、更易于理解、测试和并行化的代码。
• 代码复用与抽象：对于重复的逻辑模式，应考虑提取为泛型方法或高阶函数，以提高代码的复用性和可维护性。

在设计数据处理逻辑时，始终优先考虑使用不可变数据结构和无副作用的操作，这将显著提升代码的健壮性和可维护性。