本文探讨了在Java中处理批量数据库查询时如何通过重构代码来避免共享可变性。通过利用Java Stream API的map、flatMap和collect操作，可以消除对外部集合的副作用，从而实现更纯粹、更易于维护和并发友好的数据处理模式。
1. 批量数据处理中的共享可变性问题
在企业级应用中，从数据库批量获取数据是常见需求。然而，数据库通常对单次查询接受的参数数量有限制（例如，SQL IN 子句的参数数量）。因此，我们经常需要将一个大的键列表分割成多个小批次，然后对每个批次执行查询。
原始代码示例展示了这种场景，其中一个包含5000个数字的列表被分割成多个大小为500的子列表，然后对每个子列表执行数据库查询。
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());
Collection<list>> partitionedListOfNumbers = numbers.stream()
.collect(Collectors.groupingBy(num -> counter.getAndIncrement() / 500))
.values(); // 将列表分割成大小为500的子列表
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
});</dog></cat></list></integer></dog></cat>
上述代码的核心问题在于其使用了forEach操作，并在其中通过catList.addAll(interimCatList)和dogList.addAll(interimDogList)直接修改了外部的catList和dogList。这种模式被称为“共享可变性”（Shared Mutability）。
共享可变性带来了多方面的问题：
- 线程安全隐患： 在多线程环境下，如果多个线程同时访问并修改catList或dogList，可能导致数据不一致或运行时错误。
- 可读性和可维护性降低： 外部状态的修改使得代码难以理解，因为一个方法的行为不仅取决于其输入，还取决于其外部环境的状态。
- 不符合函数式编程范式： 函数式编程鼓励纯函数，即没有副作用的函数。修改外部状态是典型的副作用。
为了构建更健壮、更易于测试和并发友好的代码，我们应该尽量避免共享可变性。
2. 利用 Java Stream API 实现不可变数据处理
Java 8 引入的 Stream API 提供了一种声明式、函数式的方式来处理集合数据。通过利用 map、flatMap 和 collect 等操作，我们可以在不修改外部状态的情况下转换和聚合数据。
-
map： 将流中的每个元素转换成另一个元素，生成一个新的流。例如，将一个批次的键列表转换为一个List
。 -
flatMap： 将流中的每个元素转换成一个流，然后将这些流扁平化为一个单一的流。这在处理“流的流”时非常有用，例如将List
- >扁平化为List
。 - collect： 将流中的元素聚合成一个结果容器（例如List、Set或Map）。这是创建最终不可变结果的关键。
3. 重构方案详解与示例代码
为了消除共享可变性，我们将重构代码，使其不再使用forEach来修改外部列表，而是利用Stream的管道操作来生成新的结果列表。
java-api-doc-generator
下载
分析 Java Spring Boot 后端 Controller 代码，自动生成 Markdown 格式的接口文档。当用户说“生成 Java 接口文档”、“生成 API 文档”、“从 Spring 代码生成文档”时触发。
核心思路：
- 分批处理： 保持原有的分批逻辑，将大的键列表分割成多个子列表。
-
映射批次到结果： 对每个子列表，执行数据库查询，将其映射成一个结果列表（例如List
）。 -
扁平化结果列表： 将所有批次的结果列表（List
- >）扁平化成一个单一的结果流。
- 收集最终结果： 将扁平化后的结果流收集到一个新的List中。
以下是重构后的代码示例：
import java.util.ArrayList;
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.stream.Stream;
// 模拟数据库仓库接口和实体类
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 CatRepository {
public List<cat> fetchCats(List<integer> keys) {
// 模拟数据库查询
System.out.println("Fetching Cats for keys: " + keys.size() + " elements, first: " + keys.get(0));
return keys.stream()
.map(id -> new Cat(id, "Cat_" + id))
.collect(Collectors.toList());
}
}
class DogRepository {
public List<dog> fetchDogs(List<integer> keys) {
// 模拟数据库查询
System.out.println("Fetching Dogs for keys: " + keys.size() + " elements, first: " + keys.get(0));
return keys.stream()
.map(id -> new Dog(id, "Dog_" + id))
.collect(Collectors.toList());
}
}
public class BatchProcessingRefactor {
public static void main(String[] args) {
CatRepository catRepo = new CatRepository();
DogRepository dogRepo = new DogRepository();
int totalNumbers = 5000;
int batchSize = 500;
// AtomicInteger 用于在 groupingBy 中生成分组键
// 它本身是可变的，但其作用是帮助创建不可变的子集合
AtomicInteger counter = new AtomicInteger();
// 1. 数据分批：将 1 到 5000 的数字分割成大小为 500 的子列表
// IntStream.rangeClosed(1, totalNumbers) 生成一个从1到totalNumbers的整数流
// boxed() 将 IntStream 转换为 Stream<integer>
Collection<list>> partitionedListOfNumbers = IntStream.rangeClosed(1, totalNumbers)
.boxed()
.collect(Collectors.groupingBy(num -> counter.getAndIncrement() / batchSize))
.values();
System.out.println("Total partitions: " + partitionedListOfNumbers.size());
// 2. 处理 Cat 数据：使用 Stream API 避免共享可变性
// partitionedListOfNumbers.stream() 创建一个包含 List<integer> 的流
// .map(catRepo::fetchCats) 将每个 List<integer> 映射为一个 List<cat>
// 此时流的类型是 Stream<list>>
// .flatMap(List::stream) 将 Stream<list>> 扁平化为 Stream<cat>
// 即将所有 List<cat> 中的 Cat 对象合并到一个单一的流中
// .collect(Collectors.toList()) 将 Stream<cat> 中的所有 Cat 对象收集到一个新的 List<cat> 中
List<cat> catList = partitionedListOfNumbers.stream()
.map(catRepo::fetchCats)
.flatMap(List::stream)
.collect(Collectors.toList());
// 3. 处理 Dog 数据：同样的方式
List<dog> dogList = partitionedListOfNumbers.stream()
.map(dogRepo::fetchDogs)
.flatMap(List::stream)
.collect(Collectors.toList());
System.out.println("Fetched " + catList.size() + " cats.");
System.out.println("Fetched " + dogList.size() + " dogs.");
// 验证部分数据
// catList.stream().limit(5).forEach(System.out::println);
// dogList.stream().skip(4995).forEach(System.out::println);
}
}</dog></cat></cat></cat></cat></cat></list></list></cat></integer></integer></list></integer></integer></dog></integer></cat>
4. 优势与注意事项
优势：
- 避免共享可变性： catList 和 dogList 是通过 collect(Collectors.toList()) 操作创建的新列表，它们在创建过程中没有被外部修改，从而消除了副作用和线程安全问题。
- 函数式编程风格： 代码更具声明性，清晰地表达了“转换”和“聚合”的意图，而不是“迭代”和“修改”。
- 并发友好： 由于消除了共享可变状态，这种模式更容易适应并行流 (parallelStream())，从而在多核处理器上获得性能提升，而无需担心复杂的同步机制。
- 可维护性与可测试性： 减少了副作用，使得代码逻辑更纯粹，更容易进行单元测试和推理。
注意事项：
- AtomicInteger 的使用： 在 groupingBy 操作中，AtomicInteger 被用来生成不重复的分组键。虽然 AtomicInteger 本身是可变的，但它在这里的作用是辅助流操作生成不可变的子集合，而不是用于累积最终结果，因此它不会引入共享可变性问题到最终结果列表中。
- 代码重复的抽象： 示例中 catList 和 dogList 的获取逻辑存在相似性。在更复杂的场景中，可以考虑将这部分逻辑抽象为一个通用方法，接受一个函数作为参数来执行具体的数据库查询，从而进一步减少代码重复。
- 错误处理： 在实际应用中，数据库查询可能会抛出异常。Stream API 提供了 try-catch 或 Optional 等机制来处理这些情况，但需要额外设计。
5. 总结
通过采用 Java Stream API 的 map、flatMap 和 collect 等操作，我们可以有效地重构批量数据处理代码，从而避免共享可变性。这种方法不仅提升了代码的线程安全性、可读性和可维护性，还使其更符合现代函数式编程的理念。在处理大量数据或构建高并发系统时，优先考虑这种不可变的数据处理模式将带来显著的优势。
大量免费API接口：立即使用
涵盖生活服务API、金融科技API、企业工商API、等相关的API接口服务。免费API接口可安全、合规地连接上下游，为数据API应用能力赋能！
