引言：共享可变性带来的挑战
在实际的软件开发中，我们经常会遇到需要从数据库中批量获取数据的场景。然而，数据库通常会对单次查询的参数数量有所限制（例如，SQL IN子句的参数数量上限）。这意味着当我们需要查询大量ID对应的数据时，必须将这些ID列表进行分批处理，然后多次执行查询。
考虑以下Java代码示例，它试图解决分批查询的问题：
AtomicInteger counter = new AtomicInteger();
List<Cat> catList = new ArrayList<>();
List<Dog> dogList = new ArrayList<>();
List<Integer> numbers = Stream.iterate(1, e -> e + 1)
.limit(5000)
.collect(Collectors.toList());
// 将大列表分割成大小为500的小批次
Collection<List<Integer>> partitionedListOfNumbers = numbers.stream()
.collect(Collectors.groupingBy(num -> counter.getAndIncrement() / 500))
.values();
// 遍历每个批次并累加结果
partitionedListOfNumbers.stream()
.forEach(list -> {
List<Cat> interimCatList = catRepo.fetchCats(list); // 从数据库获取Cat
catList.addAll(interimCatList); // 修改外部的catList
List<Dog> interimDogList = dogRepo.fetchDogs(list); // 从数据库获取Dog
dogList.addAll(interimDogList); // 修改外部的dogList
});上述代码虽然实现了分批查询的功能，但存在一个关键问题：共享可变性（Shared Mutability）。在forEach循环内部，catList和dogList这两个外部列表被反复修改（通过addAll操作）。这种模式在单线程环境下可能不明显，但在多线程或并发环境中，可能会导致数据不一致、竞态条件等难以调试的错误。函数式编程鼓励使用不可变数据和无副作用的操作，以提高代码的健壮性和可预测性。
Java Stream API：函数式编程的解决方案
Java 8引入的Stream API提供了一种声明式、函数式的方式来处理集合数据，它强调数据流的转换而非直接修改。通过利用Stream API的map和flatMap操作，我们可以重构上述代码，彻底避免共享可变性问题。
立即学习“Java免费学习笔记（深入）”；
核心思想是：让每个批次查询操作返回其自身的结果列表，然后将所有这些结果列表“展平”并收集到一个全新的、不可变的结果列表中。
重构步骤与代码示例
为了更好地演示，我们首先定义一些辅助类和接口：
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;
// 模拟数据库查询接口
interface CatRepo {
List<Cat> fetchCats(List<Integer> keys);
}
interface DogRepo {
List<Dog> fetchDogs(List<Integer> keys);
}
// 示例实体类
class Cat {
int id;
String name;
public Cat(int id) { this.id = id; this.name = "Cat-" + id; }
@Override public String toString() { return "Cat{id=" + id + ", name='" + name + "'}"; }
}
class Dog {
int id;
String name;
public Dog(int id) { this.id = id; this.name = "Dog-" + id; }
@Override public String toString() { return "Dog{id=" + id + ", name='" + name + "'}"; }
}
// 模拟数据库查询实现
class MockCatRepo implements CatRepo {
@Override
public List<Cat> fetchCats(List<Integer> keys) {
// 模拟数据库延迟
// try { Thread.sleep(10); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
return keys.stream().map(Cat::new).collect(Collectors.toList());
}
}
class MockDogRepo implements DogRepo {
@Override
public List<Dog> fetchDogs(List<Integer> keys) {
// 模拟数据库延迟
// try { Thread.sleep(10); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
return keys.stream().map(Dog::new).collect(Collectors.toList());
}
}现在，我们使用Stream API来重构核心逻辑。
1. 生成并分批处理键列表
首先，我们需要生成一个包含所有待查询ID的列表，并将其分割成多个小批次。这里我们使用IntStream.rangeClosed生成整数序列，并结合Collectors.groupingBy进行分批。
// 初始化模拟仓库
CatRepo catRepo = new MockCatRepo();
DogRepo dogRepo = new MockDogRepo();
AtomicInteger counter = new AtomicInteger();
int BATCH_SIZE = 500;
int TOTAL_NUMBERS = 5000; // 模拟5000个ID
// 生成1到5000的整数，并按BATCH_SIZE进行分批
Collection<List<Integer>> partitionedListOfNumbers = IntStream.rangeClosed(1, TOTAL_NUMBERS)
.boxed() // 将IntStream转换为Stream<Integer>
.collect(Collectors.groupingBy(num -> counter.getAndIncrement() / BATCH_SIZE))
.values(); // 获取所有分批后的子列表集合在这个步骤中，AtomicInteger在这里的作用是为每个元素生成一个递增的索引，然后通过index / BATCH_SIZE将元素分配到不同的组中，从而实现批次划分。
2. 使用map和flatMap进行数据获取与合并
现在，我们可以利用Stream API的强大功能来并行或顺序地处理这些批次，并以不可变的方式收集结果。
- map操作： 对partitionedListOfNumbers中的每一个List<Integer>（即一个批次）调用fetchCats或fetchDogs方法。map操作会返回一个Stream<List<Cat>>或Stream<List<Dog>>，即一个包含多个列表的流。
- flatMap操作： flatMap用于将Stream<List<T>>展平为Stream<T>。它会将流中的每个内部列表展开，并将其所有元素合并到一个新的单一流中。
- collect(Collectors.toList())： 最后，将展平后的流中的所有元素收集到一个新的List中。这个新的列表是不可变的，因为它是在所有操作完成后一次性构建的。
// 获取Cat列表
List<Cat> catList = partitionedListOfNumbers.stream()
.map(catRepo::fetchCats) // 对每个批次调用fetchCats，得到 Stream<List<Cat>>
.flatMap(Collection::stream) // 将 Stream<List<Cat>> 展平为 Stream<Cat>
.collect(Collectors.toList());// 收集到最终的 List<Cat>
// 获取Dog列表
List<Dog> dogList = partitionedListOfNumbers.stream()
.map(dogRepo::fetchDogs) // 对每个批次调用fetchDogs，得到 Stream<List<Dog>>
.flatMap(Collection::stream) // 将 Stream<List<Dog>> 展平为 Stream<Dog>
.collect(Collectors.toList());// 收集到最终的 List<Dog>
// 打印结果（可选）
// System.out.println("Fetched Cats: " + catList.size());
// System.out.println("Fetched Dogs: " + dogList.size());
// System.out.println("First Cat: " + (catList.isEmpty() ? "N/A" : catList.get(0)));
// System.out.println("Last Cat: " + (catList.isEmpty() ? "N/A" : catList.get(catList.size() - 1)));通过这种方式，我们完全避免了对外部可变列表的直接修改。catList和dogList在声明时即被初始化，并在所有数据处理完成后，通过collect操作一次性赋值，确保了其不可变性。
优化：抽象通用查询逻辑
观察上述代码，catList和dogList的生成逻辑高度相似，唯一的区别在于调用的fetch方法。我们可以进一步抽象出一个通用方法来减少代码重复：
/**
* 通用方法：根据分批的键列表和查询函数，批量获取数据并合并。
* @param partitionedKeys 分批的键列表
* @param fetchFunction 每个批次对应的查询函数（例如：catRepo::fetchCats）
* @param <T> 返回结果的类型
* @return 包含所有查询结果的合并列表
*/
public static <T> List<T> fetchAnimalsInBatches(
Collection<List<Integer>> partitionedKeys,
Function<List<Integer>, List<T>> fetchFunction) {
return partitionedKeys.stream()
.map(fetchFunction) // 应用传入的查询函数
.flatMap(Collection::stream) // 展平结果
.collect(Collectors.toList());// 收集到最终列表
}
// 如何使用这个通用方法
List<Cat> catListOptimized = fetchAnimalsInBatches(partitionedListOfNumbers, catRepo::fetchCats);
List<Dog> dogListOptimized = fetchAnimalsInBatches(partitionedListOfNumbers, dogRepo::fetchDogs);
// System.out.println("Optimized Fetched Cats: " + catListOptimized.size());
// System.out.println("Optimized Fetched Dogs: " + dogListOptimized.size());这个通用方法极大地提高了代码的复用性和可维护性，使得我们可以用更简洁的方式处理不同类型的数据查询。
注意事项与最佳实践
- 不可变性与线程安全： 这种基于Stream API的方案天然地避免了共享可变性，使得代码在多线程环境下更加安全，不易出现竞态条件。
- 资源管理： Stream API本身不直接管理数据库连接等外部资源。fetchCats和fetchDogs内部的数据库操作仍需遵循标准的资源管理（如使用try-with-resources）。
- 错误处理： 数据库查询方法（如fetchCats）内部应包含适当的错误处理逻辑。如果查询可能抛出受检异常，map操作需要进行相应的处理（例如，通过包装成RuntimeException或使用Try monad等）。
- 性能考量： 对于极大规模的数据集，Stream操作可能引入一定的内存或CPU开销。但对于常见的数据库分批查询场景，其性能通常是可接受且高效的。如果需要最大化性能，可以考虑使用parallelStream()，但这会增加并发复杂性，并需要确保fetchFunction是线程安全的。
- 分批策略： 示例中使用AtomicInteger和groupingBy进行分批，这是一种有效的手段。此外，也可以使用如Guava库中的Lists.partition方法来更简洁地实现列表分批。
- 可读性： 函数式编程风格的代码通常更简洁、意图更明确，提高了代码的可读性。
总结
通过本教程，我们学习了如何利用Java Stream API的map和flatMap操作，以函数式、不可变的方式解决分批数据库查询中的共享可变性问题。这种方法不仅提升了代码的线程安全性，还使得代码更加简洁、可读性更强。在处理集合数据时，拥抱Stream API和函数式编程范式，能够帮助我们编写出更健壮、更易于维护的Java应用程序。
