Java Stream API：重构批量数据处理以避免共享可变性

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> 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 等操作，我们可以在不修改外部状态的情况下转换和聚合数据。

Cogram

使用AI帮你做会议笔记，跟踪行动项目

下载

• map： 将流中的每个元素转换成另一个元素，生成一个新的流。例如，将一个批次的键列表转换为一个List。
• flatMap： 将流中的每个元素转换成一个流，然后将这些流扁平化为一个单一的流。这在处理“流的流”时非常有用，例如将List>扁平化为List。
• collect： 将流中的元素聚合成一个结果容器（例如List、Set或Map）。这是创建最终不可变结果的关键。

3. 重构方案详解与示例代码

为了消除共享可变性，我们将重构代码，使其不再使用forEach来修改外部列表，而是利用Stream的管道操作来生成新的结果列表。

核心思路：

1. 分批处理： 保持原有的分批逻辑，将大的键列表分割成多个子列表。
2. 映射批次到结果： 对每个子列表，执行数据库查询，将其映射成一个结果列表（例如List）。
3. 扁平化结果列表： 将所有批次的结果列表（List>）扁平化成一个单一的结果流。
4. 收集最终结果： 将扁平化后的结果流收集到一个新的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> 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>
        // .flatMap(List::stream) 将 Stream> 扁平化为 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 等操作，我们可以有效地重构批量数据处理代码，从而避免共享可变性。这种方法不仅提升了代码的线程安全性、可读性和可维护性，还使其更符合现代函数式编程的理念。在处理大量数据或构建高并发系统时，优先考虑这种不可变的数据处理模式将带来显著的优势。