【工具开发】Java实现数据库数据同步

ElephantF

542人浏览 · 2025-07-08 09:30:31

ElephantF · 2025-07-08 09:30:31 发布

数据库同步数据工具

任务描述：
近期公司搞了一个Doris测试环境，用于项目上线前的测试。但是测试环境中没有任何数据，因此如果要项目测试需要将正式环境中的数据
同步到测试环境。

在开发此工具之前，同步数据是根据项目需求，将需要同步的表，在Doris正式环境中手动建表，再通过dass平台同步数据。

简单来说，同步的步骤是：

拿到前端、后端需要的表名
在Doris正式环境中获取表的DDL
在测试环境中创建表
在Dass平台同步数据

通过此工具可简化步骤123。可以某些表的步骤4(数据量小的表，10MB以下)。
接下来介绍核心代码和使用方法。

瓶颈、优化分析

初步认为瓶颈在于网络IO,Dass平台的JVM也开到8G同步是很快的。由于同步程序跑在本地16G内存的笔记本上,JVM应该是可以开到8G的。
那么瓶颈就在于程序的读写瓶颈。读写都是通过网络IO流，笔记本通过无线WIFI连接到的源服务器和目标服务器，受限于WIFI的带宽，
读写速度都没有很快。
使用有线网代替无线WIFI(待尝试)
使用Java多线程同步(待尝试)

导出数据库所有表的DDL

import java.sql.*;

public class GetTablesDDLExampleMain {

    public static void main(String[] args) throws SQLException {
        // 要同步的数据库名
        String dataBaseName = "dwd";
        Connection connection = DriverManager.getConnection(Config.getSrcUrl(dataBaseName), Config.SRC_USER, Config.SRC_PASSWORD);
        TableUtils.getDataBaseTablesDDL(connection,dataBaseName);
    }
}

同步表数据使用样例

import java.sql.*;
import java.util.List;

public class DorisDataMigratorMain {

    public static void main(String[] args) {
        // 要同步的数据库名、表名
        String dataBaseName = Table2Sync.dataBaseName;
        List<String> tables = Table2Sync.TABLES_TO_SYNC;

        System.out.println("即将同步"+ dataBaseName + "以下表：" + tables);
        try {
            // 获取数据库连接
            Connection sourceConn = DriverManager.getConnection(Config.getSrcUrl(dataBaseName), Config.SRC_USER, Config.SRC_PASSWORD);
            Connection targetConn = DriverManager.getConnection(Config.getTargetUrl(dataBaseName), Config.TARGET_USER, Config.TARGET_PASSWORD);

            // 处理表
            for (String tableName : tables) {
                if (!TableUtils.tableExists(sourceConn, tableName)) {
                    System.err.println("源数据库中不存在表：" + tableName);
                    continue;
                }
                if (!TableUtils.tableExists(targetConn, tableName)) {
                    System.err.println("目标数据库中不存在表：" + tableName);
                    continue;
                }
                System.out.println("正在迁移表：" + tableName);

                // 清空目标表
                // truncateTable(targetConn, tableName);

                // 开始迁移数据
                TableUtils.migrateTableData(sourceConn, targetConn, tableName);
            }
        }
        catch (SQLException e) {
            e.printStackTrace();
        }
    }


}

Table2Sync

用于配置需要同步的数据库和表

import java.util.Arrays;
import java.util.List;

public class Table2Sync {

    // 制定要同步的数据库
    public static final String dataBaseName = "databaseName";

    // 指定需要迁移的表名列表
    public static final List<String> TABLES_TO_SYNC = Arrays.asList(
            "table1",
            "table2",
            "table3",
            "table4",
            "table5
    );
}

Config类

主要用于配置数据源(ip,port,数据库用户名,密码)和目标数据源(ip,port,数据库用户名,密码)

public class Config {

    // 单次批量导入数据量
    public static final int BATCT_SIZE = 300;

    // 源数据库配置信息(Doris正式环境)
    public static final String SRC_USER = "src_username";
    public static final String SRC_PASSWORD = "src_passwd";
    public static String srcHost = "1.2.3.4";
    public static String srcPort = "1111"; // MySQL 协议端口


    // 目标数据库配置信息(Doris测试环境)
    public static final String TARGET_USER = "target_username";
    public static final String TARGET_PASSWORD = "target_passwd";
    public static final String targetHost = "5.6.7.8";
    public static final String targetPort = "2222";


    // 源库URL
    public static String getSrcUrl(String dataBaseName) {
        return "jdbc:mysql://" + srcHost + ":" + srcPort + "/" + dataBaseName + "?useSSL=false&characterEncoding=UTF-8";
    }

    // 目标库URL
    public static String getTargetUrl(String dataBaseName) {
        return "jdbc:mysql://" + targetHost + ":" + targetPort + "/" + dataBaseName + "?useSSL=false&characterEncoding=UTF-8";
    }
}

TableUtils类

核心方法

// 迁移表数据
public static void migrateTableData(Connection sourceConn, Connection targetConn, String tableName) throws SQLException
//  检查表是否存在
public static boolean tableExists(Connection conn, String tableName) throws SQLException
// 清空目标表数据
public static void truncateTable(Connection conn, String tableName) throws SQLException
// 导出数据库中所有表的建表语句（DDL）
public static void getDataBaseTablesDDL(Connection conn, String dataBaseName)

import java.io.BufferedWriter;
import java.io.FileWriter;
import java.sql.*;

public class TableUtils {


    // 从源表读取数据，插入到目标表中
    public static void migrateTableData(Connection sourceConn, Connection targetConn, String tableName) throws SQLException {
        String selectSql = "SELECT * FROM " + tableName;

        try (
                Statement sourceStmt = sourceConn.createStatement();
                ResultSet rs = sourceStmt.executeQuery(selectSql)
        ) {
            ResultSetMetaData metaData = rs.getMetaData();
            int columnCount = metaData.getColumnCount();

            // 构造 INSERT SQL（显式列出字段更安全，这里用 ? 占位符）
            StringBuilder placeholders = new StringBuilder();
            for (int i = 1; i <= columnCount; i++) {
                placeholders.append("?");
                if (i < columnCount) placeholders.append(", ");
            }

            String insertSql = "INSERT INTO " + tableName + " VALUES (" + placeholders.toString() + ")";
            try (PreparedStatement ps = targetConn.prepareStatement(insertSql)) {

                int count = 0;
                while (rs.next()) {
                    for (int i = 1; i <= columnCount; i++) {
                        ps.setObject(i, rs.getObject(i)); // 👉 从源表拿数据
                    }
                    ps.addBatch(); // 添加到批量插入
                    if (++count % Config.BATCT_SIZE == 0) { // 每1000条提交一次（避免内存溢出）
                        ps.executeBatch();
                        System.out.println("已插入 " + count + " 条记录");
                    }
                }
                ps.executeBatch(); // 提交剩余的数据
                System.out.println("迁移完成，共插入 " + count + " 条记录");
            }
        }
    }

    // 检查表是否存在
    public static boolean tableExists(Connection conn, String tableName) throws SQLException {
        DatabaseMetaData metaData = conn.getMetaData();
        ResultSet rs = metaData.getTables(null, null, tableName, new String[]{"TABLE"});
        boolean exists = rs.next();
        rs.close();
        return exists;
    }

    // 清空目标表
    public static void truncateTable(Connection conn, String tableName) throws SQLException {
        String sql = "TRUNCATE TABLE " + tableName;
        try (Statement stmt = conn.createStatement()) {
            stmt.executeUpdate(sql);
            System.out.println("已清空目标表：" + tableName);
        }
    }


    // 导出数据库中所有表的建表语句
    // 注意:导出的建表语句需要手动加上数据库名称
    public static void getDataBaseTablesDDL(Connection conn, String dataBaseName) {
        try{
            Statement stmt = conn.createStatement();
            ResultSet tables = stmt.executeQuery("SHOW TABLES");

            BufferedWriter writer = new BufferedWriter(new FileWriter("./SQL/" +dataBaseName + ".sql"));

            while (tables.next()) {
                String tableName = tables.getString(1);
                String showCreateSql = "SHOW CREATE TABLE " + dataBaseName + "." + tableName;

                Statement createStmt = conn.createStatement();
                ResultSet rs = createStmt.executeQuery(showCreateSql);

                if (rs.next()) {
                    String createTableSql = rs.getString(2);
                    System.out.println("-- 表: " + tableName);
                    System.out.println(createTableSql + ";\n");
                    writer.write("-- 表: " + tableName);
                    writer.newLine();
                    writer.write(createTableSql + ";");
                    writer.newLine();
                    writer.newLine();
                }
                rs.close();
                createStmt.close();
            }
            writer.close();
            System.out.println("建表语句已导出至" + dataBaseName + ".sql");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}