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

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<Integer>> 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
    });

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<Integer>> 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<Cat>>
        // .flatMap(List::stream) 将 Stream<List<Cat>> 扁平化为 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);
    }
}