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());

// 将大列表分割成大小为500的小批次
Collection<List<Integer>> partitionedListOfNumbers = numbers.stream()
    .collect(Collectors.groupingBy(num -> counter.getAndIncrement() / 500))
    .values();

// 遍历每个批次并累加结果
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.*;
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<Cat> fetchCats(List<Integer> keys);
}

interface DogRepo {
    List<Dog> fetchDogs(List<Integer> 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<Cat> fetchCats(List<Integer> 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<Dog> fetchDogs(List<Integer> keys) {
        // 模拟数据库延迟
        // try { Thread.sleep(10); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
        return keys.stream().map(Dog::new).collect(Collectors.toList());
    }
}

// 初始化模拟仓库
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<List<Integer>> partitionedListOfNumbers = IntStream.rangeClosed(1, TOTAL_NUMBERS)
    .boxed() // 将IntStream转换为Stream<Integer>
    .collect(Collectors.groupingBy(num -> counter.getAndIncrement() / BATCH_SIZE))
    .values(); // 获取所有分批后的子列表集合

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

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

// 打印结果（可选）
// 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)));

/**
 * 通用方法：根据分批的键列表和查询函数，批量获取数据并合并。
 * @param partitionedKeys 分批的键列表
 * @param fetchFunction 每个批次对应的查询函数（例如：catRepo::fetchCats）
 * @param <T> 返回结果的类型
 * @return 包含所有查询结果的合并列表
 */
public static <T> List<T> fetchAnimalsInBatches(
        Collection<List<Integer>> partitionedKeys,
        Function<List<Integer>, List<T>> fetchFunction) {
    return partitionedKeys.stream()
        .map(fetchFunction)          // 应用传入的查询函数
        .flatMap(Collection::stream) // 展平结果
        .collect(Collectors.toList());// 收集到最终列表
}

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

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