Java Stream API实践：避免共享可变性实现分批数据查询

本文探讨了在Java中处理分批数据库查询时如何避免共享可变性问题。通过利用Java Stream API的map和flatMap操作，可以以函数式、不可变的方式收集数据，从而提升代码的线程安全性和可读性，尤其适用于需要将大列表拆分为小批次进行处理的场景。

引言：共享可变性带来的挑战

在实际的软件开发中，我们经常会遇到需要从数据库中批量获取数据的场景。然而，数据库通常会对单次查询的参数数量有所限制（例如，SQL IN子句的参数数量上限）。这意味着当我们需要查询大量ID对应的数据时，必须将这些ID列表进行分批处理，然后多次执行查询。

考虑以下Java代码示例，它试图解决分批查询的问题：

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）。在forEach循环内部，catList和dogList这两个外部列表被反复修改（通过addAll操作）。这种模式在单线程环境下可能不明显，但在多线程或并发环境中，可能会导致数据不一致、竞态条件等难以调试的错误。函数式编程鼓励使用不可变数据和无副作用的操作，以提高代码的健壮性和可预测性。

Java Stream API：函数式编程的解决方案

Java 8引入的Stream API提供了一种声明式、函数式的方式来处理集合数据，它强调数据流的转换而非直接修改。通过利用Stream API的map和flatMap操作，我们可以重构上述代码，彻底避免共享可变性问题。

核心思想是：让每个批次查询操作返回其自身的结果列表，然后将所有这些结果列表“展平”并收集到一个全新的、不可变的结果列表中。

重构步骤与代码示例

为了更好地演示，我们首先定义一些辅助类和接口：

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 fetchCats(List keys);
}

interface DogRepo {
    List fetchDogs(List 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 fetchCats(List 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 fetchDogs(List 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> partitionedListOfNumbers = IntStream.rangeClosed(1, TOTAL_NUMBERS)
    .boxed() // 将IntStream转换为Stream
    .collect(Collectors.groupingBy(num -> counter.getAndIncrement() / BATCH_SIZE))
    .values(); // 获取所有分批后的子列表集合

在这个步骤中，AtomicInteger在这里的作用是为每个元素生成一个递增的索引，然后通过index / BATCH_SIZE将元素分配到不同的组中，从而实现批次划分。

2. 使用map和flatMap进行数据获取与合并

现在，我们可以利用Stream API的强大功能来并行或顺序地处理这些批次，并以不可变的方式收集结果。

• map操作： 对partitionedLis

  

  tOfNumbers中的每一个List（即一个批次）调用fetchCats或fetchDogs方法。map操作会返回一个Stream>或Stream>，即一个包含多个列表的流。
• flatMap操作： flatMap用于将Stream>展平为Stream。它会将流中的每个内部列表展开，并将其所有元素合并到一个新的单一流中。
• collect(Collectors.toList())： 最后，将展平后的流中的所有元素收集到一个新的List中。这个新的列表是不可变的，因为它是在所有操作完成后一次性构建的。

// 获取Cat列表
List catList = partitionedListOfNumbers.stream()
    .map(catRepo::fetchCats)      // 对每个批次调用fetchCats，得到 Stream>
    .flatMap(Collection::stream)  // 将 Stream> 展平为 Stream
    .collect(Collectors.toList());// 收集到最终的 List

// 获取Dog列表
List dogList = partitionedListOfNumbers.stream()
    .map(dogRepo::fetchDogs)      // 对每个批次调用fetchDogs，得到 Stream>
    .flatMap(Collection::stream)  // 将 Stream> 展平为 Stream
    .collect(Collectors.toList());// 收集到最终的 List

// 打印结果（可选）
// 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  返回结果的类型
 * @return 包含所有查询结果的合并列表
 */
public static  List fetchAnimalsInBatches(
        Collection> partitionedKeys,
        Function, List> fetchFunction) {
    return partitionedKeys.stream()
        .map(fetchFunction)          // 应用传入的查询函数
        .flatMap(Collection::stream) // 展平结果
        .collect(Collectors.toList());// 收集到最终列表
}

// 如何使用这个通用方法
List catListOptimized = fetchAnimalsInBatches(partitionedListOfNumbers, catRepo::fetchCats);
List dogListOptimized = fetchAnimalsInBatches(partitionedListOfNumbers, dogRepo::fetchDogs);

// System.out.println("Optimized Fetched Cats: " + catListOptimized.size());
// System.out.println("Optimized Fetched Dogs: " + dogListOptimized.size());

这个通用方法极大地提高了代码的复用性和可维护性，使得我们可以用更简洁的方式处理不同类型的数据查询。

注意事项与最佳实践

1. 不可变性与线程安全： 这种基于Stream API的方案天然地避免了共享可变性，使得代码在多线程环境下更加安全，不易出现竞态条件。
2. 资源管理： Stream API本身不直接管理数据库连接等外部资源。fetchCats和fetchDogs内部的数据库操作仍需遵循标准的资源管理（如使用try-with-resources）。
3. 错误处理： 数据库查询方法（如fetchCats）内部应包含适当的错误处理逻辑。如果查询可能抛出受检异常，map操作需要进行相应的处理（例如，通过包装成RuntimeException或使用Try monad等）。
4. 性能考量： 对于极大规模的数据集，Stream操作可能引入一定的内存或CPU开销。但对于常见的数据库分批查询场景，其性能通常是可接受且高效的。如果需要最大化性能，可以考虑使用parallelStream()，但这会增加并发复杂性，并需要确保fetchFunction是线程安全的。
5. 分批策略： 示例中使用AtomicInteger和groupingBy进行分批，这是一种有效的手段。此外，也可以使用如Guava库中的Lists.partition方法来更简洁地实现列表分批。
6. 可读性： 函数式编程风格的代码通常更简洁、意图更明确，提高了代码的可读性。

总结

通过本教程，我们学习了如何利用Java Stream API的map和flatMap操作，以函数式、不可变的方式解决分批数据库查询中的共享可变性问题。这种方法不仅提升了代码的线程安全性，还使得代码更加简洁、可读性更强。在处理集合数据时，拥抱Stream API和函数式编程范式，能够帮助我们编写出更健壮、更易于维护的Java应用程序。