如何在JDBC批量操作中实现异常处理与事务回滚？

原始问题与代码 #

我有一段JDBC批量插入数据的代码，希望对其进行评审，并了解如何在执行出现故障时添加事务回滚步骤。我担忧若循环过程中因某些原因抛出SQLException，该代码无法按照ACID原则回滚所有变更。

原始代码如下：

/**
 * 批量向指定表插入数据
 *
 * @param requestData 请求数据
 */
private void insertingRowsByBatches(RequestData requestData) {
    try (
            Connection connection = myDataSource.getConnection();
            Statement deleteStatement = connection.createStatement()
    ) {
        connection.setAutoCommit(TRUE);
        String stagingDeleteSql = buildMessage("DELETE FROM my_table");
        int rowsAffected = deleteStatement.executeUpdate(stagingDeleteSql);
        log.info("[{}] 条记录已删除", rowsAffected);

        String stagingInsertSql = "INSERT INTO my_table(SOME_DATA) values(?)";
        Lists.partition(Optional.ofNullable(requestData.getData()).orElse(emptyList()), MAX_ROWS_PER_INSERT)
                .forEach(recordIds -> {
                    try (PreparedStatement pstmt = connection.prepareStatement(stagingInsertSql, RETURN_GENERATED_KEYS)) {
                        for (String recordId: recordIds ) {
                            pstmt.setString(1, recordId);
                            pstmt.addBatch();
                        }
                        long start = currentTimeMillis();
                        pstmt.executeBatch();
                        long end = currentTimeMillis();
                        log.info("插入 [{}] 条数据耗时: {}ms", recordIds.size(), (end - start));
                    } catch (SQLException ex) {
                        log.error("批量插入任务失败，异常信息: {}", ex.getMessage());
                    }
                });
        connection.commit();
    } catch (Exception e) {
        throw new GenericRuntimeException(e);
    }
}

原始代码问题分析 #

• 事务配置无效：手动设置connection.setAutoCommit(TRUE)会让每条SQL执行后自动提交，完全破坏了事务的原子性，后续的connection.commit()也不会产生事务效果。
• 异常处理逻辑错误：循环内部捕获SQLException仅打印日志，未中断流程或触发回滚。这会导致部分批次插入失败后，后续批次仍继续执行，已完成的删除和插入操作无法回滚，违反ACID原则。
• 资源浪费：每次循环都创建新的PreparedStatement，不符合批量操作的性能优化原则，额外消耗数据库连接资源。

优化后的代码 #

参考相关建议后，我编写了优化后的代码，如下所示：

/**
 * 批量向指定表插入数据
 *
 * @param requestData 请求数据
 */
public void insertingRowsByBatches(RequestData requestData) {
    try (Connection connection = myDataSource.getConnection()) {
            loadInStagingTable(requestData, connection);
    } catch (SQLException e) {
        throw new GenericRuntimeException(e);
    }
}

/**
 * 将给定数据插入临时表
 *
 * @param requestData 请求数据
 * @param connection 数据库连接实例 {@link Connection}
 * @throws SQLException 数据库操作异常
 */
private void loadInStagingTable(RequestData requestData, Connection connection) throws SQLException {
    try (
            PreparedStatement deleteStatement = connection.prepareStatement("DELETE FROM my_table");
            PreparedStatement insertStatement = connection.prepareStatement("INSERT INTO my_table(SOME_DATA) values(?)", RETURN_GENERATED_KEYS)
    ) {
        connection.setAutoCommit(false);
        log.info("正在删除临时表中已存在的记录...");
        deleteStatement.executeUpdate();

        log.info("正在批量插入给定记录到临时表...");
        long start = currentTimeMillis();
        List<String> records = requestData.getData();
        List<List<String>> partitions = partition(records , MAX_ROWS_PER_INSERT);
        int partitionCount = 0;
        for (List<String> recordIds: partitions) {
            log.info("批次 [{}/{}] - 正在插入 [{}] 条记录到临时表", ++partitionCount, partitions.size(), recordIds.size());
            for (String recordId: recordIds) {
                insertStatement.setString(1, recordId);
                insertStatement.addBatch();
            }
            insertStatement.executeBatch();
        }
        connection.commit();
        log.debug("插入 [{}] 条数据总耗时: {}ms", records.size(), (currentTimeMillis() - start));
    } catch (SQLException e) {
        connection.rollback();
        throw new GenericRuntimeException(e);
    }
}

优化点说明 #

• 事务配置正确：设置connection.setAutoCommit(false)关闭自动提交，确保删除和所有插入操作处于同一个事务中，保障原子性。
• 异常回滚机制生效：将异常抛出到外层catch块，执行connection.rollback()，任何步骤失败时，已执行的所有操作都会被回滚，符合ACID原则。
• 资源复用：仅创建一次插入用的PreparedStatement并循环复用，提升性能并减少资源消耗。
• 代码解耦：将核心逻辑拆分到loadInStagingTable方法，提升代码可读性与可维护性。

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