Java JDBC百万级数据同步优化咨询：从TABLE_1到TABLE_2

针对你提供的百万级数据同步场景，原代码单条处理的方式会导致性能极低（频繁创建Statement、单条提交事务），以下是具体优化方案：

核心优化点 #

• 批量操作替代单条执行：将单条插入/删除改为批量提交，减少数据库IO次数
• 复用PreparedStatement：避免每次操作都创建新的Statement对象，降低对象创建与销毁开销
• 适配数据库的FetchSize配置：确保ResultSet真正按批次拉取数据，避免一次性加载百万条数据导致内存溢出
• 批量事务提交：每处理一定数量的数据后统一提交事务，减少事务日志的频繁写入
• 优化结果集字段读取：通过列索引替代列名读取数据，提升字段查找效率
• 完善异常处理：避免空catch吞异常，增加日志记录便于问题排查

优化后的代码示例 #

// 单线程场景下复用PreparedStatement（多线程需调整为线程私有）
private PreparedStatement insertPs;
private PreparedStatement deletePs;
private static final int BATCH_SIZE = 1000; // 批量大小可根据数据库性能调整，建议500-2000
private int batchCount = 0;

public void processData(Connection conn) throws SQLException {
    // 提前初始化批量操作的PreparedStatement
    initBatchStatements(conn);
    
    try (PreparedStatement queryPs = conn.prepareStatement("SELECT STATE, NAME, ID FROM TABLE_1")) {
        // 适配不同数据库的FetchSize：MySQL需配合URL参数useCursorFetch=true，Oracle设置为Integer.MIN_VALUE
        queryPs.setFetchSize(1000);
        
        try (ResultSet rs = queryPs.executeQuery()) {
            // 关闭自动提交，改为手动批量提交事务
            conn.setAutoCommit(false);
            
            while (rs.next()) {
                // 通过列索引读取字段，比列名查找更快
                String state = rs.getString(1);
                String name = rs.getString(2);
                String id = rs.getString(3);
                
                if ("DELETE".equals(state)) {
                    addDeleteBatch(id);
                } else if ("NEW".equals(state) || "UPDATE".equals(state)) {
                    addInsertBatch(name, id);
                }
                
                // 达到批量阈值时执行提交
                if (++batchCount % BATCH_SIZE == 0) {
                    executeBatchAndCommit(conn);
                }
            }
            
            // 处理最后一批不足BATCH_SIZE的数据
            if (batchCount % BATCH_SIZE != 0) {
                executeBatchAndCommit(conn);
            }
            
            // 恢复自动提交（可选，根据业务场景调整）
            conn.setAutoCommit(true);
        } finally {
            // 关闭复用的PreparedStatement
            closeBatchStatements();
        }
    } catch (SQLException e) {
        // 替换为实际日志框架（如SLF4J）记录异常
        System.err.println("数据处理失败: " + e.getMessage());
        // 异常时回滚事务
        if (!conn.getAutoCommit()) {
            conn.rollback();
        }
        throw e;
    }
}

private void initBatchStatements(Connection conn) throws SQLException {
    insertPs = conn.prepareStatement("INSERT INTO TABLE_2 (NAME, ID) VALUES (?, ?)");
    deletePs = conn.prepareStatement("DELETE FROM TABLE_2 WHERE ID = ?");
}

private void addInsertBatch(String name, String id) throws SQLException {
    insertPs.setString(1, name);
    insertPs.setString(2, id);
    insertPs.addBatch();
}

private void addDeleteBatch(String id) throws SQLException {
    deletePs.setString(1, id);
    deletePs.addBatch();
}

private void executeBatchAndCommit(Connection conn) throws SQLException {
    // 执行批量插入
    insertPs.executeBatch();
    // 执行批量删除
    deletePs.executeBatch();
    // 提交事务
    conn.commit();
    // 清空批量缓存
    insertPs.clearBatch();
    deletePs.clearBatch();
}

private void closeBatchStatements() throws SQLException {
    if (insertPs != null) {
        insertPs.close();
    }
    if (deletePs != null) {
        deletePs.close();
    }
}

额外注意事项 #

• 数据库参数适配：MySQL需要在JDBC URL中添加useCursorFetch=true才能使FetchSize生效；Oracle设置queryPs.setFetchSize(Integer.MIN_VALUE)会启用行预取，避免一次性加载全量数据到内存。
• 批量大小调整：BATCH_SIZE不要过大，否则可能导致数据库锁竞争加剧；也不要过小，无法发挥批量操作的性能优势。
• 线程安全：如果是多线程场景，不要全局复用PreparedStatement，建议每个线程单独初始化，或使用连接池提供的Statement缓存功能。
• 数据一致性：异常处理中必须添加事务回滚逻辑，避免出现部分数据提交、部分未提交的不一致情况。

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