1. 问题背景与共享可变性挑战
在实际的软件开发中，我们经常需要从数据库中获取大量数据。然而，数据库通常对单个查询中允许的参数数量有限制（例如，sql in 子句可能限制在500个参数）。这意味着我们不能一次性查询所有数据，而需要将大的查询列表分割成多个小批次进行查询。
考虑以下场景：我们需要根据一个包含5000个数字的列表，分批从数据库中获取猫（Cat）和狗（Dog）的信息，每次查询最多接受500个参数。一个常见的初始实现方式可能如下：
// 假设 Cat 和 Dog 是实体类，catRepo 和 dogRepo 是数据访问层接口
// CatRepo 和 DogRepo 都有 fetchCats(List<Integer>) 和 fetchDogs(List<Integer>) 方法
AtomicInteger counter = new AtomicInteger();
List<Cat> catList = new ArrayList<>(); // 共享的可变列表
List<Dog> dogList = new ArrayList<>(); // 共享的可变列表
// 模拟生成5000个数字作为查询键
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(); // 得到 List<List<Integer>> 结构
// 遍历分批的列表，执行查询并累加结果
partitionedListOfNumbers.stream()
.forEach(list -> {
List<Cat> interimCatList = catRepo.fetchCats(list); // 从数据库获取猫列表
catList.addAll(interimCatList); // 修改外部的 catList
List<Dog> 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 操作组合实现。
以下是重构后的代码示例：
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<Integer>
Collection<List<Integer>> partitionedListOfNumbers = IntStream.rangeClosed(1, 5000)
.boxed()
.collect(Collectors.groupingBy(num -> counter.getAndIncrement() / 500))
.values(); // 得到 List<List<Integer>> 结构
// 获取所有猫列表
List<Cat> catList = partitionedListOfNumbers.stream()
.map(list -> catRepo.fetchCats(list)) // 对每个分批列表执行查询，得到 Stream<List<Cat>>
.flatMap(List::stream) // 将 Stream<List<Cat>> 扁平化为 Stream<Cat>
.collect(Collectors.toList()); // 收集所有 Cat 对象到一个新的 List 中
// 获取所有狗列表
List<Dog> dogList = partitionedListOfNumbers.stream()
.map(list -> dogRepo.fetchDogs(list)) // 对每个分批列表执行查询，得到 Stream<List<Dog>>
.flatMap(List::stream) // 将 Stream<List<Dog>> 扁平化为 Stream<Dog>
.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<Cat> fetchCats(List<Integer> ids) { return ids.stream().map(id -> new Cat(id, "Cat-" + id)).collect(Collectors.toList()); } }
class DogRepo { public List<Dog> fetchDogs(List<Integer> 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();
// }代码解析：
数据分批 (groupingBy)： IntStream.rangeClosed(1, 5000).boxed().collect(Collectors.groupingBy(num -> counter.getAndIncrement() / 500)).values() 这一步与原始代码类似，负责将连续的数字列表分割成多个子列表，每个子列表包含500个数字。counter 在这里作为分组键的一部分，确保每500个数字被分到同一个组。
map 操作： partitionedListOfNumbers.stream().map(list -> catRepo.fetchCats(list)) 这一步将 Collection<List<Integer>> 转换成 Stream<List<Cat>>。对于 partitionedListOfNumbers 中的每一个 List<Integer>（即一个批次的查询键），catRepo.fetchCats(list) 会被调用，返回一个 List<Cat>。map 操作的输出是一个包含多个 List<Cat> 的流。
flatMap 操作： .flatMap(List::stream) 由于上一步 map 操作的输出是 Stream<List<Cat>>，我们希望得到的是一个单一的 List<Cat>。flatMap 的作用就是将流中的每个元素（这里是 List<Cat>）“扁平化”成一个流（通过 List::stream 方法），然后将所有这些子流连接成一个单一的流。最终，我们得到的是一个 Stream<Cat>。
collect 操作： .collect(Collectors.toList()) 这是流操作的终结操作，它将 Stream<Cat> 中的所有 Cat 对象收集到一个新的 List<Cat> 中。这个 List<Cat> 是一个全新的列表，不与任何外部变量共享，从而避免了共享可变性。
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<K> 并返回 List<T> 的查询函数
* @param <T> 结果列表中的元素类型
* @param <K> 查询键列表中的元素类型
* @return 聚合后的所有 T 类型元素的列表
*/
public <T, K> List<T> fetchEntitiesInBatches(Collection<List<K>> partitionedKeys, Function<List<K>, List<T>> fetchFunction) {
return partitionedKeys.stream()
.map(fetchFunction) // 对每个批次应用查询函数
.flatMap(List::stream) // 扁平化结果
.collect(Collectors.toList()); // 收集到新列表
}
public void fetchDataAndProcessOptimized() {
AtomicInteger counter = new AtomicInteger();
Collection<List<Integer>> partitionedListOfNumbers = IntStream.rangeClosed(1, 5000)
.boxed()
.collect(Collectors.groupingBy(num -> counter.getAndIncrement() / 500))
.values();
// 使用通用方法获取猫列表
List<Cat> catList = fetchEntitiesInBatches(partitionedListOfNumbers, catRepo::fetchCats);
// 使用通用方法获取狗列表
List<Dog> 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 操作是解决这类问题的强大工具。它们允许我们以声明式的方式对数据进行一系列无副作用的转换，最终生成新的集合。
- 拥抱函数式编程：通过避免副作用和共享可变性，我们能够编写出更纯净、更易于理解、测试和并行化的代码。
- 代码复用与抽象：对于重复的逻辑模式，应考虑提取为泛型方法或高阶函数，以提高代码的复用性和可维护性。
在设计数据处理逻辑时，始终优先考虑使用不可变数据结构和无副作用的操作，这将显著提升代码的健壮性和可维护性。
