广告：

信号屏蔽器:守护你的数字生活，为你带来宁静与专注

https://sz-sstx.com

只需要设置两个数据库ip，端口，登陆账号，密码，即可自动完成数据迁移

1.演示效果

2.主要执行代码，直接调用DatabaseMigrationJob的exe()方法即可

package com.hax.util.datasource;


import lombok.extern.slf4j.Slf4j;

import java.sql.*;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

/**
 *  mysql数据库迁移工具
 *  设置远程数据库ip，port，username，password
 *  设置目标数据库ip，port，username，password
 *  暂时适用于varchar和long类型主键，不影响迁移，只会影响性能
 *  remote指的是需要复制的数据库源
 *  goal指的是需要复制到哪个数据库
 */
@Slf4j
public class DatabaseMigrationJob {
    public static int cutThreadNum = 1000000;
    public static int batchInsertNum = 1000;
    //远程数据库信息
    public static   String remoteIp = "需要复制的数据库ip";
    public static   String remotePort = "3306";
    public static   String remoteDataBaseName = "signalblocker";
    public static   String remoteUserName = "root";
    public static   String remotePassword = "zp730990..";
    public static   Connection remoteConnection = null;
    public static   Statement remoteStatement = null;



    //----------------------------------------------------------------------------//


    //需要迁移的数据库信息

    public static String goalIp = "需要迁移到的数据库ip";//"10.6.135.12";
    public static String goalPort = "3306";
    public static String goalDataBaseName = "signalblocker";
    public static String goalUserName = "root";
    public static String goalPassword =  "zp730990..";//"rd123456,";
    public static Connection goalConnection = null;
    public static  Statement goalStatement = null;


    public DatabaseMigrationJob(){
        createDatabase(goalDataBaseName);
        System.out.println("数据库创建完毕!");
        String remoteUrl = "jdbc:mysql://"+remoteIp+":"+remotePort+"/"+remoteDataBaseName+"?useUnicode=true&characterEncoding=utf8&useSSL=true&requireSSL=true&trustCertificateKeyStoreUrl=file:/ApsaraDB-CA-Chain.jks&trustCertificateKeyStorePassword=apsaradb&serverTimezone=GMT%2B8&useOldAliasMetadataBehavior=true&allowMultiQueries=true"; // 替换为您的MySQL数据库URL
        try {
            remoteConnection = DriverManager.getConnection(remoteUrl, remoteUserName, remotePassword);
            // 创建Statement对象
            remoteStatement = remoteConnection.createStatement();
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
        String goalUrl = "jdbc:mysql://"+goalIp+":"+goalPort+"/"+goalDataBaseName+"?useUnicode=true&characterEncoding=utf8&useSSL=true&requireSSL=true&trustCertificateKeyStoreUrl=file:/ApsaraDB-CA-Chain.jks&trustCertificateKeyStorePassword=apsaradb&serverTimezone=GMT%2B8&useOldAliasMetadataBehavior=true&allowMultiQueries=true";
        try {
            goalConnection = DriverManager.getConnection(goalUrl, goalUserName, goalPassword);
            // 创建Statement对象
            goalStatement = goalConnection.createStatement();
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
    }
    public static void exe(){
        DatabaseMigrationJob databaseMigrationJob = new DatabaseMigrationJob();
        List<String> tableNames = new ArrayList<>();
        try {
            ResultSet resultSet = remoteStatement.executeQuery("SELECT table_name  \n" +
                    "FROM information_schema.tables  \n" +
                    "WHERE table_schema = '"+remoteDataBaseName+"';");
            while (resultSet.next()){
                System.out.println(resultSet.getString(1));
                tableNames.add(resultSet.getString(1));
            }
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
        databaseMigrationJob.copyTable(tableNames);
        System.out.println("表结构复制完毕!");
        databaseMigrationJob.copyData(tableNames);
        System.out.println("表数据迁移完毕！！！");
        //关闭所有连接
        databaseMigrationJob.closeAll();
    }
    public static void main(String[] args) {
        exe();
    }
    /**
     * 获取建表sql
     * @param tableName 表名称
     * @return 建表sql
     */
    public  String getTableSql(String tableName) {
        StringBuffer sql = new StringBuffer();
            ResultSet resultSet = null;
            try {
                // 执行自定义SQL查询
                String sqlQuery = "SHOW CREATE TABLE "+tableName; // 替换为您的自定义SQL查询
                try {
                    resultSet = remoteStatement.executeQuery(sqlQuery);
                } catch (SQLException e) {
                    System.out.println("获取建表语句出错:"+e.getMessage());
                    e.printStackTrace();
                }
                // 处理查询结果
                while (resultSet.next()) {
                    String createTableStatement = resultSet.getString(2);
                    return  createTableStatement;
                }
            } catch (SQLException e) {
                e.printStackTrace();
            } finally {
                // 关闭连接和资源
                try {
                    if (resultSet != null) {
                        resultSet.close();
                    }
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        return sql.toString();
    }
    /**
     * 执行指定sql
     */
    public  void exeSql(List<String> sqlList){
        for (String sql : sqlList) {
            try {
                goalStatement.execute(sql);
            } catch (SQLException e) {
                e.printStackTrace();
                System.out.println(sql);
                System.out.println("执行建表语句出错:"+e.getMessage());
            }
        }
    }
    /**
     * 迁移表结构
     */
    public  void copyTable(List<String> tableNames){
        //所有表数据建表sql
        List<String> createTableSqls = new ArrayList<>();
        //获取所有表生成sql
        for (String tableName : tableNames) {
            String tableSql = getTableSql(tableName);
            if(tableSql!=null){
                //建表语句表名去引号
                String before = tableSql.split("\\(")[0];
                String createTableLast = tableSql.replace(before,"");
                String table = tableSql.split("CREATE TABLE")[1].split("\\(")[0];
                table = table.replaceAll("`","");
                String sql = "CREATE TABLE IF NOT EXISTS" + table.toLowerCase() + " "+createTableLast;
                createTableSqls.add(sql);
            }else{
                System.out.println("实体类没有映射数据库，获取建表语句失败:");
            }
        }
        System.out.println("开始执行建表sql..................");
        //执行所有生成sql到指定服务器
        exeSql(createTableSqls);
        System.out.println("执行完毕");
    }
    /**
     * 迁移数据
     */
    public  void copyData(List<String> tableNames){
        //排除不需要迁移的文件
        List<String> excludeTables = Arrays.asList("DM_RES_POLICY_KIND".toLowerCase(),
                "DM_RES_REINS_POLICY_KIND".toLowerCase(),
                "DM_RES_POLICY_PREMIUM".toLowerCase(),
                "DM_RES_POLICY_EXPIRE_PREM".toLowerCase(),
                "DM_RES_POLICY_UNEXPIRE_PREM".toLowerCase(),
                "DM_RES_POLICY_REALPREMIUM".toLowerCase(),
                "DM_RES_CLM_PAID".toLowerCase(),
                "DM_RES_CLM_OS".toLowerCase(),
                "DM_RES_CLM_NUM".toLowerCase(),
                "DM_RES_PAID_TAX".toLowerCase(),
                "list_clm_settled".toLowerCase(),
                "dw_clm_paid_detail_kind".toLowerCase(),
                "list_clm_unsettled".toLowerCase(),
                "dw_clm_os_detail_kind".toLowerCase(),
                "list_reins_bill_slip".toLowerCase(),
                "list_reins_final_slip".toLowerCase()
        );
        tableNames = tableNames.stream().filter(t->
                !excludeTables.contains(t)
                ).collect(Collectors.toList());
        List<List<String>> lists = cuttingOriginList(tableNames, 10);
        // 创建线程池
        ExecutorService executor = Executors.newFixedThreadPool(lists.size());
        // 提交任务给线程池
        for (List<String> tableName : lists) {
            Runnable job = new DataJob(tableName);
            executor.execute(job);
        }
        System.out.println("所有数据迁移完毕！！！");
        // 关闭线程池
        executor.shutdown();
    }

    /**
     * 切割集合
     * @param data 需要切割的集合
     * @return 切割后集合的集合
     */
    public static List<List<String>> cuttingOriginList(List<String> data,int batchSize){
        List<List<String>> result = new ArrayList<>();
        if (data.size()<batchSize) {
            result.add(data);
            return result;
        }else {
            int num = 0;
            List<String> oneBatch = new ArrayList<>();
            for (int i = 0; i < data.size(); i++) {
                num++;
                oneBatch.add(data.get(i));
                if(num%batchSize==0){
                    result.add(oneBatch);
                    oneBatch = new ArrayList<>();
                }else if(num==data.size()){
                    result.add(oneBatch);
                    oneBatch = new ArrayList<>();
                }
            }
            return result;
        }
    }

    /**
     * 关闭数据库所有连接
     */
    public void closeAll(){
        try {
            remoteConnection.close();
            remoteStatement.close();
            goalConnection.close();
            goalStatement.close();
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 创建指定名称数据库
     * @param databaseName 数据库名称
     */
    public void createDatabase(String databaseName){
        Connection connection;
        Statement statement;
        try {
            connection = DriverManager.getConnection("jdbc:mysql://"+goalIp+":"+goalPort,goalUserName,goalPassword);
            statement = connection.createStatement();
            String createDbSql = "CREATE DATABASE IF NOT EXISTS `"+databaseName+"`";
            statement.execute(createDbSql);
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }

    }
}

3.内部调用线程类

package com.hax.util.datasource;

import lombok.extern.slf4j.Slf4j;

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

@Slf4j
public class DataJob implements Runnable {

    //批量提交的数量
    private  int batchInsertNum = DatabaseMigrationJob.batchInsertNum;
    //远程数据库信息
    private  String remoteIp = DatabaseMigrationJob.remoteIp;
    private  String remotePort = DatabaseMigrationJob.remotePort;
    private  String remoteDataBaseName = DatabaseMigrationJob.remoteDataBaseName;
    private  String remoteUserName = DatabaseMigrationJob.remoteUserName;
    private  String remotePassword = DatabaseMigrationJob.remotePassword;
    private  Connection remoteConnection = null;
    private  Statement remoteStatement = null;



    //----------------------------------------------------------------------------//


    //需要迁移的数据库信息

    private  String goalIp = DatabaseMigrationJob.goalIp;
    private  String goalPort = DatabaseMigrationJob.goalPort;
    private  String goalDataBaseName = DatabaseMigrationJob.goalDataBaseName;
    private  String goalUserName = DatabaseMigrationJob.goalUserName;
    private  String goalPassword = DatabaseMigrationJob.goalPassword;
    private  Connection goalConnection = null;
    private   Statement goalStatement = null;
    private List<String> tableNames = null;
    public DataJob(List<String> tableNames){
        this.tableNames = tableNames;
        String remoteUrl = "jdbc:mysql://"+remoteIp+":"+remotePort+"/"+remoteDataBaseName+"?useUnicode=true&characterEncoding=utf8&useSSL=true&requireSSL=true&trustCertificateKeyStoreUrl=file:/ApsaraDB-CA-Chain.jks&trustCertificateKeyStorePassword=apsaradb&serverTimezone=GMT%2B8&useOldAliasMetadataBehavior=true&allowMultiQueries=true"; // 替换为您的MySQL数据库URL
        try {
            remoteConnection = DriverManager.getConnection(remoteUrl, remoteUserName, remotePassword);
            // 创建Statement对象
            remoteStatement = remoteConnection.createStatement();
        } catch (SQLException e) {
            log.error("远程数据库连接失败");
        }
        String goalUrl = "jdbc:mysql://"+goalIp+":"+goalPort+"/"+goalDataBaseName+"?useUnicode=true&characterEncoding=utf8&useSSL=true&requireSSL=true&trustCertificateKeyStoreUrl=file:/ApsaraDB-CA-Chain.jks&trustCertificateKeyStorePassword=apsaradb&serverTimezone=GMT%2B8&useOldAliasMetadataBehavior=true&allowMultiQueries=true"; // 替换为您的MySQL数据库URL
        try {
            goalConnection = DriverManager.getConnection(goalUrl, goalUserName, goalPassword);
            // 创建Statement对象
            goalStatement = goalConnection.createStatement();
        } catch (SQLException e) {
            log.error("目标数据源连接失败");
        }
    }
    @Override
    public void run() {
        try {
            for (String tableName : tableNames) {
                copyTableData(tableName);
                System.out.println(tableName+"数据迁移完成");
            }
        }catch (Exception e){
            System.out.println("出现错误");
            e.printStackTrace();
            try {
                remoteConnection.close();
                remoteStatement.close();
                goalConnection.close();
                goalStatement.close();
            }catch (Exception exception){

            }
        }
    }


    /**
     * 执行某个表的数据迁移
     */
    public  void copyTableData(String tableName){
            //判断改表是否已经成功迁移
        //查询已经迁移完的数据部分数量
        int num = isSuccess(tableName);
        if (num<0) {
            return;
        }
            ResultSet rs = null;
            try {
                int index = num;
                int next ;
                while (true){
                    next = index+100000;
                    // 执行自定义SQL查询
                    String sqlQuery = "SELECT SQL_NO_CACHE  * FROM  "+tableName+" LIMIT "+index+","+next; // 替换为您的自定义SQL查询
                    rs = remoteStatement.executeQuery(sqlQuery);
                    List<List<Object>> dataList = new ArrayList<>();
                    while (rs.next()) {
                        ResultSetMetaData metaData = rs.getMetaData();
                        int columnCount = metaData.getColumnCount();
                        List<Object> rowData = new ArrayList<>();
                        for (int i = 1; i <= columnCount; i++) {
                            rowData.add(rs.getObject(i));
                        }
                        dataList.add(rowData);
                    }
                    List<List<List<Object>>> lists = cuttingList(dataList,batchInsertNum);
                    for (List<List<Object>> list : lists) {
                        for (List<Object> rowData : list) {
                            //获取主键字段名称，并判断表主键是否是自增
                            DatabaseMetaData metaData = goalConnection.getMetaData();
                            ResultSet resultSet = metaData.getColumns(null, null, tableName,null);
                            //主键名称
                            String keyName = null;
                            boolean isAutoIncrement = false;
                            while (resultSet.next()){
                                keyName = resultSet.getString("COLUMN_NAME");
                                isAutoIncrement = resultSet.getBoolean("IS_AUTOINCREMENT");
                                // 判断主键是否是自增的
                                if (isAutoIncrement) {
                                    System.out.println(keyName + " 是自增主键");
                                }
                                break;
                            }

                            Object o = rowData.get(0);
                            String key = "";
                            //查询数据库是否存在相同主键数据
                            System.out.println(tableName);
                            ResultSet keyData = null;
                            String sql = "";
                            if(o instanceof String){
                                key = (String) rowData.get(0);
                                sql = "select * from "+tableName+" where "+keyName+" = '" +key+"'";
                            }else if(o instanceof Long){
                                key = String.valueOf((Long)rowData.get(0));
                                sql = "select * from "+tableName+" where "+keyName+" = " +key;
                            }else if(o instanceof Integer){
                                key = String.valueOf(rowData.get(0)) ;
                                sql = "select * from "+tableName+" where "+keyName+" = " +key;
                            }
                            try {
                                keyData = goalStatement.executeQuery(sql);
                            } catch (SQLException e) {
                                e.printStackTrace();
                                System.out.println(sql);
                            }
                            try {
                                if(keyData!=null&&keyData.isBeforeFirst()&&keyData.next()){
                                    System.out.println("主键重复:"+key);
                                    continue;
                                }
                            }catch (SQLException e){
                                System.out.println("结果集异常");
                                continue;
                            }

                            StringBuffer insertQuery = new StringBuffer("INSERT INTO "+tableName+" VALUES (");
                            //自增主键第一个值不需要
                            if (isAutoIncrement) {
                                rowData.set(0,null);
                            }
                            for (int i = 0; i < rowData.size(); i++) {
                                if (i > 0) {
                                    insertQuery.append(", ");
                                }
                                if (rowData.get(i) instanceof String) {
                                    //判断字符串是"开头开始'开头
                                    int i1 = ((String) rowData.get(i)).indexOf("\"");
                                    int i2 = ((String) rowData.get(i)).indexOf("'");
                                    if(i2==-1||i1<i2){
                                        insertQuery.append("'").append(rowData.get(i)).append("'");
                                    }else {
                                        insertQuery.append("\"").append(rowData.get(i)).append("\"");
                                    }

                                } else {
                                    if(rowData.get(i)==null){
                                        insertQuery.append(rowData.get(i));
                                    }else {
                                        int i1 = (String.valueOf(rowData.get(i))).indexOf("\"");
                                        int i2 = (String.valueOf(rowData.get(i))).indexOf("'");
                                        if(i2==-1||i1<i2){
                                            insertQuery.append("'").append(rowData.get(i)).append("'");
                                        }else {
                                            insertQuery.append("\"").append(rowData.get(i)).append("\"");
                                        }
                                    }
                                }
                            }
                            insertQuery.append(")");
//                            System.out.println(insertQuery);
                            try {
                                goalStatement.addBatch(insertQuery.toString());
                            }catch (Exception e){
                                e.printStackTrace();
                                System.out.println("addBatch出错:"+e.getMessage());
                            }
                        }
                        try {
                            goalStatement.executeBatch();
                        } catch (SQLException e) {
                            e.printStackTrace();
                            System.out.println("执行出错:"+e.getMessage());
                        }
                    }
                    if(dataList.size()<100000){
                        break;
                    }
                    index+=100000;
                }

            }
            catch (SQLException e) {
                e.printStackTrace();
            } finally {
                // 关闭连接和资源
                try {
                    if (rs != null) {
                        rs.close();
                    }
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
    }

    /**
     * 判断该表是否已经迁移完成
     * @param tableName
     * @return 如果目标库表数据量大于等于远程库表数据量返回-1 ；
     *         否则返回目标库已经存在的数据量
     */
    public int isSuccess(String tableName){
        int result = 0;
        ResultSet remoteResult = null;
        ResultSet goalResult = null;
        int remoteCount = 0;
        int goalCount = 0;
        try {
            remoteResult = remoteStatement.executeQuery("select count(*) from  " + tableName);
            if(remoteResult.next()){
               remoteCount  = remoteResult.getInt(1);
            }
            goalResult = goalStatement.executeQuery("select count(*) from  " + tableName);
            if(goalResult.next()){
                goalCount = goalResult.getInt(1);
            }
            result = goalCount - remoteCount>=0?-1:goalCount;
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }finally {
            try {
                if(remoteResult!=null){
                    remoteResult.close();
                }
                if(goalResult!=null){
                    goalResult.close();
                }
            } catch (SQLException e) {
                System.out.println(tableName);
                throw new RuntimeException(e);
            }
        }
        return result;
    }



    /**
     * 切割集合
     * @param data 需要切割的集合
     * @return 切割后集合的集合
     */
    public  List<List<List<Object>>> cuttingList(List<List<Object>> data,int batchSize){
        List<List<List<Object>>> result = new ArrayList<>();
        if (data.size()<=batchSize) {
            result.add(data);
            return result;
        }else {
            int start = 0;
            List<List<Object>> oneBatch = new ArrayList<>();
            for (int i = 0; i < data.size(); i++) {
                start++;
                oneBatch.add(data.get(i));
                if(start>=batchSize){
                    result.add(oneBatch);
                    oneBatch = new ArrayList<>();
                    start = 0;
                }
            }
            return result;
        }
    }




}

更多问题请在sz-sstx.com平台的产品与解决方案留言!