java多线程批量导入数据库中的数据有哪些

Java多线程批量导入数据库中的数据详解

在现代软件开发中，数据批量导入是一个常见的需求，无论是处理大量的日志文件、配置文件还是从其他系统迁移数据，高效的批量导入都显得尤为重要，Java作为一个强大的编程语言，提供了多种方式来实现多线程批量导入数据库中的数据，本文将详细介绍几种常用的方法,并通过代码示例来展示它们的实现过程。

我们来看一下为什么需要使用多线程进行批量导入，单线程的批量导入虽然简单直观，但在面对大量数据时，其性能往往不尽如人意，尤其是在I/O密集型操作（如数据库写入）的情况下，多线程可以充分利用CPU资源,提高整体的处理效率。

使用ExecutorService管理线程池

ExecutorService是Java并发包中的一个接口，它提供了一个灵活的方式来管理线程池，通过创建一个固定大小的线程池，我们可以有效地控制并发执行的任务数量,避免过多的线程导致系统资源耗尽。

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class MultiThreadedBatchImport {
    private static final int BATCH_SIZE = 100;
    private static final String DB_URL = "jdbc:mysql://localhost:3306/yourDatabase";
    private static final String USER = "username";
    private static final String PASS = "password";
    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(10); // 创建包含10个线程的线程池
        for (int i = 0; i < 1000; i++) {
            final int id = i;
            executor.submit(() -> {
                // 模拟从外部获取数据的过程
                String data = "data" + id;
                // 插入数据到数据库
                insertData(data);
            });
        }
        executor.shutdown();
    }
    private static void insertData(String data) {
        Connection conn = null;
        PreparedStatement pstmt = null;
        try {
            conn = DriverManager.getConnection(DB_URL, USER, PASS);
            String sql = "INSERT INTO yourTable (columnName) VALUES (?)";
            pstmt = conn.prepareStatement(sql);
            pstmt.setString(1, data);
            pstmt.addBatch();
            // 达到批处理大小后执行
            if (pstmt.getBatchSize() == BATCH_SIZE) {
                pstmt.executeBatch();
                pstmt.clearBatch();
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (pstmt != null) pstmt.close();
                if (conn != null) conn.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

使用ForkJoin框架
ForkJoinPool是Java 7引入的一个高级并发框架，它提供了一种分而治之的策略来处理大任务，通过递归地分割任务，并在多个线程上并行执行子任务,最终合并结果。
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.ForkJoinPool;
public class ForkJoinBatchImport extends RecursiveTask<Void> {
    private static final int BATCH_SIZE = 100;
    private static final String DB_URL = "jdbc:mysql://localhost:3306/yourDatabase";
    private static final String USER = "username";
    private static final String PASS = "password";
    private final int start;
    private final int end;
    public ForkJoinBatchImport(int start, int end) {
        this.start = start;
        this.end = end;
    }
    @Override
    protected Void compute() {
        if (end - start <= BATCH_SIZE) {
            insertDataRange(start, end);
        } else {
            int mid = (start + end) / 2;
            ForkJoinBatchImport left = new ForkJoinBatchImport(start, mid);
            ForkJoinBatchImport right = new ForkJoinBatchImport(mid, end);
            left.fork();
            right.compute();
            left.join();
        }
        return null;
    }
    private void insertDataRange(int start, int end) {
        Connection conn = null;
        PreparedStatement pstmt = null;
        try {
            conn = DriverManager.getConnection(DB_URL, USER, PASS);
            String sql = "INSERT INTO yourTable (columnName) VALUES (?)";
            pstmt = conn.prepareStatement(sql);
            for (int i = start; i < end; i++) {
                pstmt.setString(1, "data" + i);
                pstmt.addBatch();
            }
            pstmt.executeBatch();
            pstmt.clearBatch();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (pstmt != null) pstmt.close();
                if (conn != null) conn.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
    public static void main(String[] args) {
        ForkJoinPool pool = new ForkJoinPool();
        pool.invoke(new ForkJoinBatchImport(0, 1000));
    }
}

使用CompletableFuture进行异步处理
CompletableFuture是Java 8引入的一个功能强大的工具类，它允许你以非阻塞的方式编写异步代码，通过组合多个CompletableFuture实例,我们可以构建复杂的异步工作流。

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.IntStream;
public class CompletableFutureBatchImport {
    private static final int BATCH_SIZE = 100;
    private static final String DB_URL = "jdbc:mysql://localhost:3306/yourDatabase";
    private static final String USER = "username";
    private static final String PASS = "password";
    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(10); // 创建包含10个线程的线程池
        IntStream.range(0, 1000).parallel().forEach(i -> {
            CompletableFuture.runAsync(() -> {
                // 模拟从外部获取数据的过程
                String data = "data" + i;
                // 插入数据到数据库
                insertData(data);
            }, executor);
        });
    }
    private static void insertData(String data) {
        Connection conn = null;
        PreparedStatement pstmt = null;
        try {
            conn = DriverManager.getConnection(DB_URL, USER, PASS);
            String sql = "INSERT INTO yourTable (columnName) VALUES (?)";
            pstmt = conn.prepareStatement(sql);
            pstmt.setString(1, data);
            pstmt.addBatch();
            // 达到批处理大小后执行
            if (pstmt.getBatchSize() == BATCH_SIZE) {
                pstmt.executeBatch();
                pstmt.clearBatch();
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (pstmt != null) pstmt.close();
                if (conn != null) conn.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

三种方法各有优缺点，选择哪种取决于具体的需求和场景，对于简单的批量导入任务，ExecutorService可能是最直接有效的方式；而对于更复杂的任务分解和组合，ForkJoin和CompletableFuture则提供了更强大的功能。