本文探讨了在Java中处理批量数据库查询时如何通过重构代码来避免共享可变性。通过利用Java Stream API的map、flatMap和collect操作，可以消除对外部集合的副作用，从而实现更纯粹、更易于维护和并发友好的数据处理模式。
1. 批量数据处理中的共享可变性问题
在企业级应用中，从数据库批量获取数据是常见需求。然而，数据库通常对单次查询接受的参数数量有限制（例如，SQL IN 子句的参数数量）。因此，我们经常需要将一个大的键列表分割成多个小批次，然后对每个批次执行查询。
原始代码示例展示了这种场景，其中一个包含5000个数字的列表被分割成多个大小为500的子列表，然后对每个子列表执行数据库查询。
AtomicInteger counter = new AtomicInteger();List catList = new ArrayList(); // 外部可变列表List dogList = new ArrayList(); // 外部可变列表List 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 interimCatList = catRepo.fetchCats(list); // 从数据库获取Cat catList.addAll(interimCatList); // 修改外部 catList List interimDogList = dogRepo.fetchDogs(list); // 从数据库获取Dog dogList.addAll(interimDogList); // 修改外部 dogList });
上述代码的核心问题在于其使用了forEach操作，并在其中通过catList.addAll(interimCatList)和dogList.addAll(interimDogList)直接修改了外部的catList和dogList。这种模式被称为“共享可变性”（Shared Mutability）。
共享可变性带来了多方面的问题：
立即学习“Java免费学习笔记（深入）”；
线程安全隐患： 在多线程环境下，如果多个线程同时访问并修改catList或dogList，可能导致数据不一致或运行时错误。可读性和可维护性降低： 外部状态的修改使得代码难以理解，因为一个方法的行为不仅取决于其输入，还取决于其外部环境的状态。不符合函数式编程范式： 函数式编程鼓励纯函数，即没有副作用的函数。修改外部状态是典型的副作用。
为了构建更健壮、更易于测试和并发友好的代码，我们应该尽量避免共享可变性。
2. 利用 Java Stream API 实现不可变数据处理
Java 8 引入的 Stream API 提供了一种声明式、函数式的方式来处理集合数据。通过利用 map、flatMap 和 collect 等操作，我们可以在不修改外部状态的情况下转换和聚合数据。
map： 将流中的每个元素转换成另一个元素，生成一个新的流。例如，将一个批次的键列表转换为一个List。flatMap： 将流中的每个元素转换成一个流，然后将这些流扁平化为一个单一的流。这在处理“流的流”时非常有用，例如将List<List>扁平化为List。collect： 将流中的元素聚合成一个结果容器（例如List、Set或Map）。这是创建最终不可变结果的关键。
3. 重构方案详解与示例代码
为了消除共享可变性，我们将重构代码，使其不再使用forEach来修改外部列表，而是利用Stream的管道操作来生成新的结果列表。
核心思路：
分批处理： 保持原有的分批逻辑，将大的键列表分割成多个子列表。映射批次到结果： 对每个子列表，执行数据库查询，将其映射成一个结果列表（例如List）。扁平化结果列表： 将所有批次的结果列表（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 fetchCats(List 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 fetchDogs(List 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 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 的流 // .map(catRepo::fetchCats) 将每个 List 映射为一个 List // 此时流的类型是 Stream<List> // .flatMap(List::stream) 将 Stream<List> 扁平化为 Stream // 即将所有 List 中的 Cat 对象合并到一个单一的流中 // .collect(Collectors.toList()) 将 Stream 中的所有 Cat 对象收集到一个新的 List 中 List catList = partitionedListOfNumbers.stream() .map(catRepo::fetchCats) .flatMap(List::stream) .collect(Collectors.toList()); // 3. 处理 Dog 数据：同样的方式 List 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); }}
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 等操作，我们可以有效地重构批量数据处理代码，从而避免共享可变性。这种方法不仅提升了代码的线程安全性、可读性和可维护性，还使其更符合现代函数式编程的理念。在处理大量数据或构建高并发系统时，优先考虑这种不可变的数据处理模式将带来显著的优势。
以上就是Java Stream API：重构批量数据处理以避免共享可变性的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/40257.html
微信扫一扫 
支付宝扫一扫 
