java 多线程数据库插入千万级数据

在 Java 中实现多线程数据库插入千万级数据，可以采用线程池的方式来实现。

首先，创建一个线程池，然后将数据分割成若干个数据块，每个线程处理一个数据块，将数据插入到数据库中。在插入数据时，可以采用批处理的方式，将多条数据一次性插入到数据库中，以提高效率。

下面是一个简单的示例代码，用于实现多线程数据库插入千万级数据：

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class MultiThreadInsert {

    private static final String url = "jdbc:mysql://localhost:3306/test?useSSL=false&serverTimezone=UTC";
    private static final String user = "root";
    private static final String password = "root";

    public static void main(String[] args) {
        int threadNum = 10; // 线程数
        int dataSize = 10000000; // 数据量
        int batchSize = 1000; // 批处理大小

        // 数据块大小
        int blockSize = dataSize / threadNum;
        if (blockSize % batchSize != 0) {
            blockSize = (blockSize / batchSize + 1) * batchSize;
        }

        // 生成数据
        List<List<String>> dataList = generateData(dataSize);

        // 线程池
        ExecutorService executor = Executors.newFixedThreadPool(threadNum);

        // 数据库连接
        Connection conn = null;
        try {
            conn = DriverManager.getConnection(url, user, password);

            // 插入数据
            for (int i = 0; i < threadNum; i++) {
                int start = i * blockSize;
                int end = start + blockSize;
                if (end > dataSize) {
                    end = dataSize;
                }
                List<List<String>> subList = dataList.subList(start, end);
                executor.execute(new InsertTask(subList, conn, batchSize));
            }

            // 关闭线程池
            executor.shutdown();
            while (!executor.isTerminated()) {
                Thread.sleep(100);
            }

            System.out.println("数据插入完成！");
        } catch (SQLException | InterruptedException e) {
            e.printStackTrace();
        } finally {
            if (conn != null) {
                try {
                    conn.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    // 生成数据
    private static List<List<String>> generateData(int dataSize) {
        List<List<String>> dataList = new ArrayList<>();
        for (int i = 0; i < dataSize; i++) {
            List<String> row = new ArrayList<>();
            row.add("value1");
            row.add("value2");
            row.add("value3");
            dataList.add(row);
        }
        return dataList;
    }
}

// 插入数据任务
class InsertTask implements Runnable {

    private List<List<String>> dataList;
    private Connection conn;
    private int batchSize;

    public InsertTask(List<List<String>> dataList, Connection conn, int batchSize) {
        this.dataList = dataList;
        this.conn = conn;
        this.batchSize = batchSize;
    }

    @Override
    public void run() {
        try {
            String sql = "INSERT INTO test_table (column1, column2, column3) VALUES (?, ?, ?)";
            PreparedStatement ps = conn.prepareStatement(sql);

            int count = 0;
            for (List<String> row : dataList) {
                ps.setString(1, row.get(0));
                ps.setString(2, row.get(1));
                ps.setString(3, row.get(2));
                ps.addBatch();

                count++;
                if (count % batchSize == 0) {
                    ps.executeBatch();
                    count = 0;
                }
            }

            if (count > 0) {
                ps.executeBatch();
            }

            ps.close();
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }
}

在上面的示例代码中，我们使用了线程池来创建多个线程，将数据分割成若干个数据块，每个线程处理一个数据块，将数据插入到数据库中。同时，我们还使用了批处理的方式，将多条数据一次性插入到数据库中，以提高效率。