如何用ExecutorService分批处理数据

在 Java 中，ExecutorService 常用于并发处理任务。当数据量很大时，通常会把数据分批（分片）提交给线程池，避免一次创建过多任务或 OOM。下面从思路 → 示例 → 注意事项一步步说明。

一、基本思路

1. 准备数据（List / 数据库结果等）
2. 按批次拆分数据
3. 每批创建一个任务（Runnable / Callable）
4. 提交给 ExecutorService
5. 等待所有批次执行完成
6. 关闭线程池

二、简单示例：分批处理 List 数据

1. 定义分批工具方法

public static <T> List<List<T>> split(List<T> list, int batchSize) {
    List<List<T>> result = new ArrayList<>();
    for (int i = 0; i < list.size(); i += batchSize) {
        result.add(list.subList(i, Math.min(list.size(), i + batchSize)));
    }
    return result;
}

2. 使用 ExecutorService 分批处理

ExecutorService executor = Executors.newFixedThreadPool(4);

List<Integer> data = IntStream.range(0, 1000)
        .boxed()
        .collect(Collectors.toList());

int batchSize = 100;
List<List<Integer>> batches = split(data, batchSize);

List<Future<?>> futures = new ArrayList<>();

for (List<Integer> batch : batches) {
    futures.add(executor.submit(() -> {
        // 处理每一批数据
        for (Integer item : batch) {
            System.out.println(Thread.currentThread().getName() + " : " + item);
        }
    }));
}

// 等待所有任务完成
for (Future<?> future : futures) {
    future.get();
}

executor.shutdown();

三、推荐写法（更简洁：invokeAll）

ExecutorService executor = Executors.newFixedThreadPool(4);

List<Callable<Void>> tasks = new ArrayList<>();

for (List<Integer> batch : batches) {
    tasks.add(() -> {
        batch.forEach(System.out::println);
        return null;
    });
}

executor.invokeAll(tasks);
executor.shutdown();

✅ invokeAll 会阻塞直到所有任务完成

四、生产者-消费者模式（适合超大数据）

如果数据量极大（如百万级），不建议一次性拆分：

ExecutorService executor = Executors.newFixedThreadPool(4);

BlockingQueue<List<Integer>> queue = new LinkedBlockingQueue<>();

// 生产者
executor.submit(() -> {
    for (List<Integer> batch : batches) {
        queue.put(batch);
    }
});

// 消费者
for (int i = 0; i < 4; i++) {
    executor.submit(() -> {
        while (true) {
            List<Integer> batch = queue.poll(1, TimeUnit.SECONDS);
            if (batch == null) break;
            batch.forEach(System.out::println);
        }
    });
}

五、常见注意事项

✅ 线程池大小

• CPU 密集型：CPU + 1
• IO 密集型：可更大（如 2×CPU）

✅ 批大小

• 太小：线程切换开销大
• 太大：单任务耗时过长

✅ 异常处理

try {
    future.get();
} catch (ExecutionException e) {
    e.getCause().printStackTrace();
}

✅ 一定要 shutdown 否则程序不会退出

六、总结一句话

ExecutorService 分批处理 = 数据切片 + 每批一个任务 + 线程池并发执行

如果你有具体场景（如数据库、文件、Spring、CompletableFuture），我可以给你更贴合的示例。