使用Java实现每小时从旧库同步新增数据到新库的问题求助

API 探路者

2026-10-19

代码问题排查

执行方法调用错误
INSERT属于数据写入类操作，必须使用executeUpdate()执行，该方法会返回影响的行数；你当前使用的executeQuery()仅用于执行SELECT类查询，返回结果集，调用后会直接抛出SQL异常，你注释掉的那行代码才是该语句的正确调用方式。
跨库操作前提限制
你现在使用的INSERT INTO b.table2 SELECT * FROM a.table1语法，仅支持两个库在同一个MySQL实例、且当前连接用户拥有a、b两个库的读写权限的场景，如果新旧库部署在不同服务器，该语法完全不生效，需要分别建立两个数据库连接处理。
驱动兼容性问题
如果使用的是MySQL 8.0及以上版本，旧驱动类com.mysql.jdbc.Driver已经废弃，需要更换为com.mysql.cj.jdbc.Driver；且JDBC 4.0及以上版本无需手动通过Class.forName加载驱动，SPI机制会自动扫描加载。
资源泄漏风险
当前代码没有在异常场景下关闭数据库连接、Statement等资源，容易出现连接泄漏，建议使用try-with-resources语法自动释放资源。

增量同步实现方案

增量字段配置：给旧库需要同步的表选择唯一的增量标识，优先选择自增主键，次选create_time（仅新增同步）/update_time（新增+更新同步），每次同步结束后记录本次同步的最大标识值，下次同步仅拉取大于该值的部分，避免全表扫描拉取重复数据。
定时任务配置：每小时执行的需求可以直接用JDK自带的ScheduledExecutorService实现，Spring项目可直接使用@Scheduled(cron = "0 0 * * * ?")注解配置整点执行。
性能优化：拉取数据使用分页查询避免单次加载大量数据导致OOM，写入新库使用PreparedStatement的批量操作接口，大幅提升写入性能。
可靠性保证：新增同步进度持久化逻辑，同步失败后记录当前进度，下次执行自动从失败位置续传；写入新库使用INSERT ON DUPLICATE KEY UPDATE语法保证幂等性，避免重复同步数据导致主键冲突。

修正后通用同步代码示例（跨实例场景）

import java.sql.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class DBSync {
    // 旧库配置
    private static final String OLD_DB_URL = "jdbc:mysql://旧库地址/旧库名?useUnicode=true&characterEncoding=utf8";
    private static final String OLD_DB_USER = "旧库用户名";
    private static final String OLD_DB_PWD = "旧库密码";
    // 新库配置
    private static final String NEW_DB_URL = "jdbc:mysql://新库地址/新库名?useUnicode=true&characterEncoding=utf8";
    private static final String NEW_DB_USER = "新库用户名";
    private static final String NEW_DB_PWD = "新库密码";
    // 上一次同步的最大自增ID，实际使用可以持久化到配置文件/数据库
    private static long lastMaxId = 0;

    public static void sync() {
        // 使用try-with-resources自动关闭资源
        try (Connection oldCon = DriverManager.getConnection(OLD_DB_URL, OLD_DB_USER, OLD_DB_PWD);
             Connection newCon = DriverManager.getConnection(NEW_DB_URL, NEW_DB_USER, NEW_DB_PWD);
             Statement oldStmt = oldCon.createStatement();
             // 增量拉取数据
             ResultSet rs = oldStmt.executeQuery("SELECT id,col1,col2,col3 FROM 旧表名 WHERE id > " + lastMaxId);
             // 预编译插入语句，开启批量操作
             PreparedStatement newPstmt = newCon.prepareStatement("INSERT INTO 新表名(id,col1,col2,col3) VALUES (?,?,?,?) ON DUPLICATE KEY UPDATE col1=VALUES(col1),col2=VALUES(col2),col3=VALUES(col3)")
        ) {
            int batchSize = 1000;
            int count = 0;
            long currentMaxId = lastMaxId;
            while (rs.next()) {
                long id = rs.getLong(1);
                newPstmt.setLong(1, id);
                newPstmt.setString(2, rs.getString(2));
                newPstmt.setString(3, rs.getString(3));
                newPstmt.setString(4, rs.getString(4));
                newPstmt.addBatch();
                count++;
                currentMaxId = Math.max(currentMaxId, id);
                // 每1000条执行一次批量写入
                if (count % batchSize == 0) {
                    newPstmt.executeBatch();
                    newPstmt.clearBatch();
                }
            }
            // 执行剩余的批量操作
            newPstmt.executeBatch();
            // 更新同步进度
            lastMaxId = currentMaxId;
            System.out.println("本次同步完成，共同步" + count + "条数据，当前同步最大ID：" + lastMaxId);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        // 启动定时任务，每小时执行一次
        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
        scheduler.scheduleAtFixedRate(DBSync::sync, 0, 1, TimeUnit.HOURS);
    }
}

内容的提问来源于stack exchange，提问作者fifi.h