Java Stream API：重构代码以避免共享可变性

1. 问题背景与共享可变性分析

在处理大数据量查询时，数据库通常会对sql语句中的参数数量有所限制。例如，一个in子句可能只接受最多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());

// 将大列表分割成多个大小为500的子列表
Collection> partitionedListOfNumbers = numbers.stream()
    .collect(Collectors.groupingBy(num -> counter.getAndIncrement() / 500))
    .values();

// 遍历分批列表，并修改外部集合
partitionedListOfNumbers.stream()
    .forEach(list -> {
        List interimCatList = catRepo.fetchCats(list); // 从数据库获取Cat
        catList.addAll(interimCatList); // 修改外部catList
        List interimDogList = dogRepo.fetchDogs(list); // 从数据库获取Dog
        dogList.addAll(interimDogList); // 修改外部dogList
    });

上述代码的核心问题在于其使用了共享可变性（Shared Mutability）。catList和dogList是在forEach循环外部声明的，并在循环内部通过addAll方法不断被修改。这种模式在函数式编程中被视为“不纯”的操作，因为它产生了副作用（Side Effect）。共享可变性带来的弊端包括：

• 难以理解和维护： 程序的行为依赖于执行顺序和外部状态，增加了心智负担。
• 并发安全问题： 在多线程环境下，多个线程同时修改同一个共享集合可能导致数据不一致或竞态条件，需要额外的同步机制（如Collections.synchronizedList或CopyOnWriteArrayList），这会增加复杂性和开销。
• 测试困难： 带有副作用的代码更难进行单元测试，因为测试结果可能受外部状态影响。

2. 解决方案：利用Java Stream API实现无副作用聚合

Java 8引入的Stream API提供了一种声明式、函数式的数据处理方式，可以有效地避免共享可变性。通过结合map、flatMap和collect操作，我们可以在不修改任何外部状态的情况下，将分批查询的结果聚合到新的集合中。

// 更简洁的数字列表生成方式
Collection> partitionedListOfNumbers = IntStream.rangeClosed(1, 5000)
    .boxed() // 将IntStream转换为Stream
    .collect(Collectors.groupingBy(num -> (num - 1) / 500)) // 修正分组逻辑，确保从0开始计数
    .values();

// 获取Cat列表：使用Stream API进行无副作用聚合
List catList = partitionedListOfNumbers.stream()
    .map(list -> catRepo.fetchCats(list)) // 将每个小批次键列表映射为List
    .flatMap(List::stream) // 将Stream>扁平化为Stream
    .collect(Collectors.toList()); // 收集所有Cat对象到新的List

// 获取Dog列表：同样使用Stream API进行无副作用聚合
List dogList = partitionedListOfNumbers.stream()
    .map(list -> dogRepo.fetchDogs(list)) // 将每个小批次键列表映射为List
    .flatMap(List::stream) // 将Stream>扁平化为Stream
    .collect(Collectors.toList()); // 收集所有Dog对象到新的List

代码解析：

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

1. IntStream.rangeClosed(1, 5000).boxed(): 这是一个更简洁高效地生成Integer列表的方式，避免了AtomicInteger的引入，因为groupingBy的classifier函数是无状态的。boxed()将IntStream转换为Stream。
2. collect(Collectors.groupingBy(num -> (num - 1) / 500)).values(): 这一步负责将原始的数字列表按每500个一组进行分区。(num - 1) / 500作为分组键，确保每500个数字分到一个组中。
3. partitionedListOfNumbers.stream(): 创建一个包含所有分批键列表的流。
4. .map(list -> catRepo.fetchCats(list)): 这是核心转换步骤。对于流中的每个List（即一个批次的键），调用catRepo.fetchCats(list)方法从数据库中获取一个List。此时，流的类型是Stream>。
5. .flatMap(List::stream): flatMap操作用于将一个流中的每个元素（这里是List)转换成一个流，然后将所有这些子流连接成一个单一的流。List::stream是一个方法引用，它将每个List转换成一个Stream。最终，我们得到了一个Stream。
6. .collect(Collectors.toList()): 最后，使用Collectors.toList()将扁平化后的所有Cat对象收集到一个新的List中。这个新列表是不可变的，因为它是在收集操作结束时一次性构建的，没有在过程中被外部修改。

通过这种方式，catList和dogList都是通过流操作的最终结果创建的新集合，避免了在处理过程中对外部共享状态的修改。

SlidesAI

使用SlidesAI的AI在几秒钟内创建演示文稿幻灯片

下载

3. 进一步抽象与代码复用

由于获取Cat列表和Dog列表的逻辑结构非常相似，我们可以进一步抽象，减少代码重复。这可以通过创建一个通用的辅助方法来实现，该方法接受一个函数作为参数，用于指定如何从数据库获取特定类型的实体。

import java.util.Collection;
import java.util.List;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

// 假设Cat和Dog类以及catRepo, dogRepo已定义

public class AnimalService {

    // 辅助方法，用于通用地从分批的键列表中获取实体
    private static  List fetchEntities(
        Collection> partitionedKeys,
        Function, List> fetchFunction) {

        return partitionedKeys.stream()
            .map(fetchFunction) // 应用传入的获取函数
            .flatMap(List::stream)
            .collect(Collectors.toList());
    }

    public static void main(String[] args) {
        // 模拟数据仓库
        CatRepository catRepo = new CatRepository();
        DogRepository dogRepo = new DogRepository();

        // 生成并分区键列表
        Collection> partitionedListOfNumbers = IntStream.rangeClosed(1, 5000)
            .boxed()
            .collect(Collectors.groupingBy(num -> (num - 1) / 500))
            .values();

        // 使用辅助方法获取Cat列表
        List catList = fetchEntities(partitionedListOfNumbers, catRepo::fetchCats);
        System.out.println("Fetched " + catList.size() + " cats.");

        // 使用辅助方法获取Dog列表
        List dogList = fetchEntities(partitionedListOfNumbers, dogRepo::fetchDogs);
        System.out.println("Fetched " + dogList.size() + " dogs.");
    }
}

// 模拟实体类和仓库接口
class Cat {}
class Dog {}

class CatRepository {
    public List fetchCats(List keys) {
        // 模拟数据库查询
        return keys.stream().map(k -> new Cat()).collect(Collectors.toList());
    }
}

class DogRepository {
    public List fetchDogs(List keys) {
        // 模拟数据库查询
        return keys.stream().map(k -> new Dog()).collect(Collectors.toList());
    }
}

通过引入fetchEntities辅助方法，我们不仅避免了共享可变性，还提高了代码的模块化和复用性。Function, List>作为参数，使得该方法可以灵活地应用于任何接受List并返回List的数据库查询操作。

4. 总结与注意事项

通过Java Stream API的map、flatMap和collect操作，我们能够以声明式、函数式的方式重构代码，彻底避免了共享可变性问题。这种方法带来了多方面的好处：

• 纯净性与可预测性： 函数不再有副作用，其输出仅依赖于输入，提高了代码的可预测性。
• 并发安全性： 由于没有共享状态被修改，代码天然地更适合并发执行，尤其是在使用parallelStream()时。
• 可读性与维护性： 声明式代码通常更简洁，更易于理解其意图，而非关注具体的执行步骤。
• 易于测试： 无副作用的函数更容易进行单元测试，因为它们不依赖或修改外部状态。

注意事项：

• 性能考量： 尽管Stream API提供了强大的抽象，但在某些极端性能敏感的场景下，手动循环可能略快。然而，对于大多数业务应用而言，Stream API带来的代码质量提升远超微小的性能差异。
• 流的惰性求值： Stream是惰性求值的，只有当遇到终端操作（如collect）时，中间操作才会执行。
• 理解map与flatMap的区别： map将每个元素转换成一个新的元素（可能是一个集合），而flatMap则将每个元素转换成一个流，并将所有这些流扁平化为一个单一的流。这是聚合分批结果的关键。

通过采纳这些函数式编程原则和Stream API的最佳实践，开发者可以编写出更健壮、更易于维护和扩展的Java应用程序。