java 多线程处理大批量数据

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class BulkDataProcessing {

    // 模拟大批量数据处理任务
    public static void processData(int id) {
        System.out.println("Processing data batch: " + id + " by thread: " + Thread.currentThread().getName());
        try {
            // 模拟处理时间
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    public static void main(String[] args) {
        int totalBatches = 100; // 总数据批次
        int threadPoolSize = 10; // 线程池大小

        ExecutorService executorService = Executors.newFixedThreadPool(threadPoolSize);

        // 提交任务到线程池
        for (int i = 0; i < totalBatches; i++) {
            final int batchId = i;
            executorService.submit(() -> processData(batchId));
        }

        // 关闭线程池并等待所有任务完成
        executorService.shutdown();
        try {
            if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
                executorService.shutdownNow();
            }
        } catch (InterruptedException e) {
            executorService.shutdownNow();
        }

        System.out.println("All data batches processed.");
    }
}

解释说明：

1. processData(int id) 方法:

   • 模拟一个批量数据处理的任务，每个任务会打印当前处理的数据批次编号和执行该任务的线程名称。
   • 使用 Thread.sleep(1000) 模拟处理时间。

2. main 方法:

   • 定义了总共需要处理的数据批次 totalBatches 和线程池的大小 threadPoolSize。
   • 使用 Executors.newFixedThreadPool(threadPoolSize) 创建一个固定大小的线程池。
   • 使用 executorService.submit() 提交任务到线程池中，每个任务都会调用 processData 方法来处理一批数据。
   • 调用 executorService.shutdown() 关闭线程池，并使用 awaitTermination 等待所有任务完成。如果在指定时间内没有完成，则强制关闭线程池。

3. 线程池的作用:

   • 线程池可以有效地管理和复用线程，避免频繁创建和销毁线程带来的性能开销。
   • 通过控制线程池的大小，可以限制并发任务的数量，防止系统资源被耗尽。

4. 异常处理:

   • 在 processData 方法中捕获 InterruptedException 并恢复线程的中断状态。
   • 在 main 方法中捕获 InterruptedException 并确保线程池能够正确关闭。

这个示例展示了如何使用 Java 的多线程机制来高效地处理大批量数据。