Java实现MySQL跨库表查询批量插入不生效及仅插最后一条问题咨询

问题根因 #

• 原代码将ps1.executeBatch()放在while循环内部，每次循环仅添加1条数据就执行批量操作，本质和单条插入无差异，所以执行速度极慢。
• 挪到循环外仅插入最后一条的核心原因有两点：
  1. 代码存在类型匹配错误：ps1.setInt(1,rs.getString("data")) 把字符串类型的查询结果直接赋值给int类型参数，赋值异常会导致之前的批次数据失效，最终只有最后一条侥幸执行成功。
  2. MySQL JDBC驱动默认未开启批量语句重写，即使调用addBatch也会拆解为单条语句执行，部分场景下会出现数据丢失问题。
• 额外的隐患：如果table1的数据量过大，一次性把所有数据加入批次会占用过多内存，可能引发OOM。

修复方案 #

1. 修正参数类型匹配错误，确保set方法的参数类型和数据库字段、查询结果类型一致。
2. 在第一个数据库的JDBC连接URL末尾添加参数rewriteBatchedStatements=true，开启批量语句重写，真正实现多数据合并插入。
3. 调整批次执行逻辑，每积累N条（建议1000~5000条，根据数据大小调整）就执行一次批量并清空批次，避免内存占用过高。
4. 把executeBatch从循环内挪到批次触发节点，循环内仅执行addBatch操作。

修复后代码示例 #

public static void insert(){
    // 每1000条执行一次批量，可根据实际情况调整阈值
    int BATCH_SIZE = 1000;
    int x = 0;
    try{
        // 第一个数据库连接URL末尾必须加 rewriteBatchedStatements=true 才能真正生效批量插入
        Class.forName("com.mysql.jdbc.Driver");
        Connection con = DriverManager.getConnection("jdbc:mysql://xxx/xxx?rewriteBatchedStatements=true", "账号", "密码");
        // 第二个数据库连接
        Class.forName("com.mysql.cj.jdbc.Driver");
        Connection con2 = DriverManager.getConnection("jdbc:mysql://xxx/xxx", "账号", "密码");
        con.setAutoCommit(false);
        con2.setAutoCommit(false);
        PreparedStatement ps1 = con.prepareStatement("INSERT INTO table2 (data) VALUES(?)");
        PreparedStatement ps = con2.prepareStatement("SELECT * FROM table1");
        ResultSet rs = ps.executeQuery();
        while(rs.next()){
            x++;
            System.out.println(x + " added to batch");
            // 修正类型匹配问题，如果data字段是字符串就用setString，是int就用getInt，和字段类型对应即可
            ps1.setString(1,rs.getString("data"));
            ps1.addBatch();
            // 达到批次阈值就执行一次批量
            if(x % BATCH_SIZE == 0){
                ps1.executeBatch();
                ps1.clearBatch();
            }
        }
        // 执行剩余不满阈值的批次数据
        ps1.executeBatch();
        con.commit();
        con2.commit();
        // 资源关闭建议补充在finally块，此处省略
    }catch (Exception e){
        e.printStackTrace();
        try {
            con.rollback();
        } catch (Exception rollbackEx) {
            rollbackEx.printStackTrace();
        }
    }
}

内容的提问来源于stack exchange，提问作者Sparrow