Java多线程批量数据导入的方法详解

什么是多线程数据导入？

• 多线程数据导入是指在进行大量数据录入时，可以通过多个线程来同时完成数据导入工作，提高数据导入效率的一种方式。

• 在数据量较大的场景下，使用多线程能够更快地完成数据导入操作，缩短数据导入时间，提高导入数据的效率。

多线程数据导入的步骤

1. 初始化一个线程池（可控制线程数），每个线程对应一个数据处理任务。
2. 将数据划分成固定大小的数据块，使用 CountDownLatch 来实现多个线程的并发处理。
3. 启动线程池，等待所有线程都处理完成。

多线程数据导入的代码实现

public class MultiThreadImportService {
    private ExecutorService executorService;

    public MultiThreadImportService(int threadNum) {
        executorService = Executors.newFixedThreadPool(threadNum);
    }

    /*
     * data：导入数据
     * batchSize: 单个任务处理的数据量
     * task: 数据处理任务接口，需自行实现
     */
    public void importData(List<?> data, int batchSize, DataProcessTask task) {
        int dataSize = data.size();
        int totalTasks = (dataSize + batchSize - 1) / batchSize;

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

        // 控制器，多线程同时开始
        CountDownLatch latch = new CountDownLatch(totalTasks);

        for (int i = 0; i < totalTasks; i++) {
            int start = i * batchSize;
            int end = (i == totalTasks - 1) ? dataSize : (start + batchSize);

            List<?> subData = data.subList(start, end);

            // 增加一个导入任务
            MultiThreadDataTask<?> taskTemp = new MultiThreadDataTask<>(subData, task, latch);

            futures.add(executorService.submit(taskTemp));
        }

        // 等待所有任务完成
        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        // 输出所有任务的处理结果
        for (Future<?> f : futures) {
            try {
                System.out.println(f.get());
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        }

        // 关闭线程池
        executorService.shutdown();
    }
}

// 数据处理任务接口
public interface DataProcessTask {
    void process(List<?> datalist);
}

// 导入数据任务实现类
public class MultiThreadDataTask<T> implements Callable<Object> {
    private List<T> data;
    private DataProcessTask task;
    private CountDownLatch latch;

    public MultiThreadDataTask(List<T> data, DataProcessTask task, CountDownLatch latch) {
        this.data = data;
        this.task = task;
        this.latch = latch;
    }

    @Override
    public Object call() {
        try {
            task.process(data);
        } finally {
            latch.countDown();
        }

        return String.format("线程%s处理完毕，处理数据量：%d", Thread.currentThread().getName(), data.size());
    }
}

示例说明

示例一：

场景描述：有一份10000条学生信息需要导入到数据库中。

public class Student {
    private String name;
    private Integer age;
    // 其他属性省略......
}

public class StudentImportTask implements DataProcessTask {

    @Override
    public void process(List<?> datalist) {
        List<Student> students = (List<Student>) datalist;
        // 批量保存学生信息到数据库
    }
}

List<Student> students = new ArrayList<>();
// 假设这里添加了10000条学生信息

MultiThreadImportService importService = new MultiThreadImportService(10); // 创建一个线程池，线程数为10

importService.importData(students, 500, new StudentImportTask()); // 数据按500条为一组保存，调用学生信息导入任务进行并发处理

在以上示例中，我们使用了一个线程池，分成了10个线程对数据进行处理，数据按照500条为一组进行拆分，然后用 CountDownLatch 来实现这些线程的并发处理，最后输出每个任务的处理结果。

示例二：

场景描述：有一份100000条订单信息需要导入到数据库中。

public class Order {
    private String orderNo;
    private BigDecimal amount;
    // 其他属性省略......
}

public class OrderImportTask implements DataProcessTask {

    @Override
    public void process(List<?> datalist) {
        List<Order> orders = (List<Order>) datalist;
        // 批量保存订单信息到数据库
    }
}

List<Order> orders = new ArrayList<>();
// 假设这里添加了100000条订单信息

MultiThreadImportService importService = new MultiThreadImportService(20); // 创建一个线程池，线程数为20

importService.importData(orders, 1000, new OrderImportTask()); // 数据按1000条为一组保存，调用订单信息导入任务进行并发处理

在以上示例中，我们同样使用了一个线程池，分成了20个线程对数据进行处理，数据按照1000条为一组进行拆分，并用 CountDownLatch 来实现这些线程的并发处理，最后输出每个任务的处理结果。

总结

通过以上的示例，我们可以看出，在实现多线程数据导入的过程中，需要注意以下几点：

1. 初始化一个线程池，控制线程数，避免过度消耗系统资源。
2. 将数据划分为固定大小的数据块，使用 CountDownLatch 来实现多个线程的并发处理。
3. 等待所有任务都完成后输出每个任务的处理结果。

最后，需要注意的是，在使用多线程处理数据时，需要保证数据的事务性，避免数据错误与数据丢失等问题。