纯jdbc操作oracle数据库迁移至mysql数据库（测试2000张表，单表数据超过百万，不会出现OOM）

傷城~ 2023-06-30 01:36 60阅读 0赞

### 文章目录 ###

*   *  前言
     *  编码前思考
     *  实现
     *   *  demo结构
         *  引入依赖
     *  实现类
     *  优化
     *  源码

## 前言 ##

数据迁移向来是麻烦的，如果有oracle转mysql 需求，数据迁移就是一大问题  
在数据迁移前曾百度一些方案：  
1.使用nacavit的数据传输功能  
![在这里插入图片描述][20200113154953304.png]  
跑着跑着就失败了，问题难以分析

1.  使用kettle  
    kettle使用起来其实还是挺方便的，但是唯一不好的地方是每个表都要做抽数脚本，要表少，一个一个做倒还好  
    ，但是现在系统有2000张表  
    ![在这里插入图片描述][20200113155217466.png]
2.  github寻找开源项目  
    倒是找到一个阿里的开源项目：[yugong][]  
    ![在这里插入图片描述][20200113155347968.png]

因为太过专业所以使用起来也是需要时间学习成本，其次就是自己需要迁移的数据需要满足一些自己的逻辑

1.  后来想了想为什么不能自己写个抽数脚本呢？现在框架用多了，思维渐渐僵化，想了jdbc很容易实现

## 编码前思考 ##

1.  如何迁移数据  
    通过查询获取oracle数据，然后拼接insert sql语句在mysql 中执行
2.  使用什么模式  
    这里觉得生产者消费者模式特别适合，一个线程从oracle生产insert语句，一个线程在mysql执行insert语句
3.  拼接sql语句存在哪里  
    考虑线程问题这里决定使用线程安全的队列BlockingQueue

## 实现 ##

### demo结构 ###

![在这里插入图片描述][20200113160234685.png]

### 引入依赖 ###

<dependencies>
            <dependency>
                <groupId>mysql</groupId>
                <artifactId>mysql-connector-java</artifactId>
                
                <version>5.1.38</version>
            </dependency>
    
            <dependency>
                <groupId>com.oracle</groupId>
                <artifactId>ojdbc6</artifactId>
                <version>1.0</version>
                <scope>system</scope>
                <systemPath>${ project.basedir}/src/main/resources/lib/ojdbc6.jar</systemPath>
            </dependency>
        </dependencies>

> ps 这里oracle依赖使用本地依赖，因为公用maven库是无法现在收费的oracle jar，其次mysql jar 最好使用8以，不然元数据获取的是表字段是获取整个数据库而不是单个数据库

## 实现类 ##

*  Main

/** * @author wh * @version 1.0 * @date 2019-12-13 17:06 */
    public class Main { 
    
        static final String MYSQL_DRIVER = "com.mysql.jdbc.Driver";
        static final String MYSQL_URL = "jdbc:mysql://192.168.1.101:3306/test?useUnicode=true&autoReconnect=true&allowMultiQueries=true&useSSL=false&serverTimezone=Asia/Shanghai";
        static final String MYSQL_USERNAME = "root";
        static final String MYSQL_PASSWORD = "123456";
    
        static final String ORACLE_DRIVER = "oracle.jdbc.driver.OracleDriver";
        static final String ORACLE_URL = "jdbc:oracle:thin:@192.168.1.103:1601:FTSHARE";
        static final String ORACLE_USERNAME = "root";
        static final String ORACLE_PASSWORD = "123456";
    
        static Connection mysqlCon = null;
        static Connection oracleCon = null;
    
        static List<Table> tables = new ArrayList<>();
    
        static BlockingQueue<String> blockingQueue = new LinkedBlockingQueue<>(30);
    
    
        public static void main(String[] args) throws Exception { 
            try { 
                //初始化连接
                mysqlCon = init(MYSQL_DRIVER, MYSQL_URL, MYSQL_USERNAME, MYSQL_PASSWORD);
                oracleCon = init(ORACLE_DRIVER, ORACLE_URL, ORACLE_USERNAME, ORACLE_PASSWORD);
                DatabaseMetaData dbMetData = mysqlCon.getMetaData();
                //获取元数据rs
                ResultSet rs = Utils.getRs(dbMetData);
                //获取所有表名
                tables = Utils.getTables(rs);
                //获取所有表的所有列 及每个表的列总数
                tables = Utils.setAllColumns(dbMetData, tables);
                //获取oracle执行sql
                tables = Utils.generateOracleSelectSQL(tables);
                // 消费线程，插入mysql
                Consumer consumer = new Consumer(blockingQueue/*, mysqlCon, FLAG*/);
                Thread consumerThread = new Thread(consumer);
                consumerThread.start();
                //通过查询oracleSql获取mysqlSql,放入BlockingQueue
                Utils.generateMysqlInsertSQL(tables, oracleCon, blockingQueue);
            } catch (Exception e) { 
                e.printStackTrace();
            } finally { 
                if (mysqlCon != null) { 
                    try { 
                        mysqlCon.close();
                        oracleCon.close();
                    } catch (SQLException e) { 
                        e.printStackTrace();
                    }
                }
            }
    
        }
    
        /** * 初始化连接 * @param driver * @param url * @param username * @param password * @return * @throws Exception */
        public static Connection init(String driver, String url, String username, String password) throws Exception { 
            Class.forName(driver);
            System.out.println("驱动： " + driver + "获取连接成功");
            return DriverManager.getConnection(url, username, password);
        }
    }

*  Table.java

/** * @author wh * @version 1.0 * @date 2019-12-13 16:13 */
    public class Table { 
    
        /** * 表名 */
        String tablename;
        /** * 所有列,必须使用list集合，使用string直接拼接后期对关键字冲突不好拼接'' */
        List<String> columns;
        /** * 列总数 */
        int columnSum;
        /** * 表中数据总量 */
        int sum;
        /** * mysql执行sql语句 */
        String mysqSb;
        /** * oracle执行sql */
        String oracleSb;
    
    	//省略 get set方法

*  Utils

/** * @author wh * @version 1.0 * @date 2019-12-13 11:53 */
    public class Utils { 
        /** * mysql 50条数据为1条insert */
        static int BATCH = 50;
        /** * oracle 100为一页 */
        static final int PAGING = 100;
        /** * 获取查询结果总条数 * @param rs * @return int 查询结果总条数 * @throws Exception */
        public static int sumColumn(ResultSet rs) throws Exception { 
            rs.last();
            //每个表的列数
            int sumColum = rs.getRow();
            rs.beforeFirst();
            return  sumColum;
        }
    
    
    
        /** * 获取数据库所有表名 * @param rs * @return List<Table> * @throws Exception */
        public static List<Table> getTables(ResultSet rs) throws Exception{ 
            /* * ResultSet rs = dbMetData.getTables(null, Utils.convertDatabaseCharsetType("zou", "mysql"), null, new String[] { "TABLE"}); * * */
            List<Table> tables = new ArrayList<>();
            System.out.println("获取表总数为：" + Utils.sumColumn(rs));
            while (rs.next()) { 
                if (rs.getString(4) != null
                        && (rs.getString(4).equalsIgnoreCase("TABLE"))) { 
                    Table table = new Table();
    
                    String tableName = rs.getString(3).toLowerCase();            
                    table.setTablename(tableName);
                    tables.add(table);
    
                }
            }
            return tables;
    
        }
    
        /** * 获取所有表的 列 及列总数 * @param dbMetData * @param tables 所有表 * @return * @throws Exception */
        public static List<Table> setAllColumns(DatabaseMetaData dbMetData, List<Table> tables) throws Exception { 
            tables.forEach(table -> { 
                try { 
                    Table table1 = getColumn(table.getTablename(), dbMetData);
                    table.setColumns(table1.getColumns());
                    table.setColumnSum(table1.getColumnSum());
                } catch (Exception e) { 
                    e.printStackTrace();
                }
            }
            );
            return tables;
        }
    
    
        //根据表名获取 列 及列总数
        public static Table getColumn(String tableName, DatabaseMetaData dbMetData) throws Exception{ 
            Table table = new Table();
            String columnName;
            //列
            List<String> list = new ArrayList<>();
            ResultSet colRet = dbMetData.getColumns(null, "%", tableName,
                    "%");
            //获取列总数
            int sumColum = Utils.sumColumn(colRet);
    
            //获取所有列
            while (colRet.next()) { 
                columnName = colRet.getString("COLUMN_NAME");
                //去掉mysql主键,如果为主键 列总数 -1
                if (columnName.equals("ID_")){ 
                    sumColum--;
                    continue;
                }
                list.add(columnName);
            }
            table.setColumns(list);
            table.setColumnSum(sumColum);
            return  table;
        }
    
        /** * 获取oracle执行sql * @param tables * @return */
        public static List<Table> generateOracleSelectSQL(List<Table> tables) { 
            StringBuffer sb = new StringBuffer("select ");
            for (Table table : tables) { 
                for (String column : table.getColumns()) { 
                    sb.append(column + ",");
                }
                //去掉拼接sql后最后一个，
                sb = sb.deleteCharAt(sb.length()-1);
                sb.append(" from " + table.getTablename());
                table.setOracleSb(sb + "");
                //重新new一个对象 不能使用 setlength 因为指向同一个引用
                 sb = new StringBuffer("select ");
    
    
            }
            return tables;
        }
    
    
        /** * 获取元数据rs * @param dbMetData * @return * @throws Exception */
        public static ResultSet getRs(DatabaseMetaData dbMetData) throws Exception { 
            //获取元数据 rs
            ResultSet rs = dbMetData.getTables(null,
                    null, null,
                    new String[] {  "TABLE"});
            return rs;
    
        }
    
        // 获取mysql所有执行sql
        public static void generateMysqlInsertSQL(List<Table> tables, Connection oracleCon, BlockingQueue<String> blockingQueue) throws Exception { 
    
    
            ResultSet resultSet = null;
            PreparedStatement statement = null;
            //查询总数据了sql
            String sql;
            //表名
            String tableNmae;
            //表总数据量
            int sum;
    
    
            for (Table table : tables) { 
                String oracleSql = table.getOracleSb();
                //获取表名
                tableNmae = table.getTablename();
                try { 
                    /** * todo 内存溢出 * 大对象，沾满老年代 （老年代2.5G） * 1 分页 * 2 大内存 */
                    System.out.println("表: " + tableNmae  + " 开始加载数据..........");
                    //查询表数据总量
                    sum =  Utils.getsum(oracleCon, tableNmae);
                    //将表总量回插table对象中
                    table.setSum(sum);
                    //对大数据表进行分页 10000W为分页基础
                    if (sum > PAGING) { 
                        Utils.putblocking(sum, oracleCon, table, blockingQueue);
                        return;
                    }
    
                    resultSet = statement.executeQuery(oracleSql + "");
    
                    //如果表中有数据则拼接sql存放在BlockingQueue
    
                    if (table.getSum() > 0) { 
                      generateMysqlInsertSQLByTable(resultSet, table, blockingQueue);
                    }
    
    
                    // 释放 resultSet
                    if(resultSet != null) { 
                        resultSet.close();
                    }
                    if(statement != null) { 
                        statement.close();
                    }
    
                } catch (Exception e) { 
                    e.printStackTrace();
                    System.out.println("表: " + table.getTablename() + "不存在");
                }
    
    
            }
            System.out.println("所有表导出完成！！");
        }
    
        //通过rs拼接单个mysql SQL
        public static void generateMysqlInsertSQLByTable(ResultSet rs, Table table, BlockingQueue<String> queue) throws Exception { 
    
    
            //获取mysql拼接列
            String mysqlColumn = Utils.getMysqlColum(table);
    
            /** * todo 这个地方一条一条的生成sql语句 * 100条一个sql */
    
            // prefix
            StringBuffer mysqlSql = new StringBuffer("insert into " + table.getTablename() + "( " + mysqlColumn + " )" + "values(");
            int count = 0;
            while (rs.next()) { 
                count++;
                // a,b,c,d
                String singleValue = geneateOneSQL(rs, table);
    
                // insert values( + a,b,c,d
                mysqlSql.append(singleValue);
    
    
                if (count % BATCH == 0) { 
                    // insert values( + a,b,c,d + )
                    mysqlSql.append(")");
                    // 放入队列
                    queue.put(mysqlSql.toString());
                    mysqlSql = new StringBuffer("insert into " + table.getTablename() + "( " + mysqlColumn + " )" + "values(");
    
    
                } else { 
                    // insert values( + a,b,c,d + ),(
                    mysqlSql.append("),(");
                }
    
            }
    
            if (count % BATCH != 0) { 
                mysqlSql = mysqlSql.deleteCharAt(mysqlSql.length() - 1);
                mysqlSql = mysqlSql.deleteCharAt(mysqlSql.length() - 1);
    
                // 放入队列
                queue.put(mysqlSql.toString());
            }
    
    
        }
        public static String geneateOneSQL(ResultSet rs, Table table) throws  Exception { 
    
            StringBuffer singleSQL = new StringBuffer("");
    
            // 返回一条数据的values 'a','b','c','d'
            for (int i =1; i <= table.getColumnSum(); i++) { 
    
                String value = rs.getString(i);
    
                if (value == null) { 
                    //为null 不拼接 ''
                    singleSQL.append(value + ",");
                } else { 
                    if (value.contains("'")) { 
                        value = value.replace("'","''");
                    }
                    singleSQL.append("'" + value + "'" + ",");
                }
            }
    
            /** * 去掉末尾逗号 */
            singleSQL = singleSQL.deleteCharAt(singleSQL.length() - 1);
    
            return  singleSQL.toString();
    
        }
    
        /** * 获取mysql 拼接sql列 * @param table * @return */
        public static String getMysqlColum(Table table) { 
            StringBuffer sb = new StringBuffer();
            for (String colum : table.getColumns()) { 
                sb.append("`" + colum + "`,");
            }
            //去掉最后的,
            sb = sb.deleteCharAt(sb.length() - 1);
            return sb + "";
        }
    
    
        /** * 查询表总数据量 * @param connection * @param tableNmae * @return * @throws Exception */
        public static int getsum(Connection connection, String tableNmae) throws Exception{ 
            int sum = 0;
    
            String sql = "select count(1) from " + tableNmae;
            PreparedStatement ps = connection.prepareStatement(sql);
            ResultSet rs = ps.executeQuery();
            while (rs.next()) { 
                sum = rs.getInt(1);
    
            }
            System.out.println("表数据总量为: " + sum);
            ps.close();
            return sum;
    
        }
    
    
        /* * oracle分页 * * 得到分页 SQL * */
    
        public static String getResultSet(int index, int lastPage, Table table) { 
            StringBuffer sb = new StringBuffer("select ");
                for (String column : table.getColumns()) { 
                    sb.append(column + ",");
                }
                //去掉拼接sql后最后一个，
                sb = sb.deleteCharAt(sb.length()-1);
            sb.append(" FROM (SELECT ROWNUM AS rowno, t.* FROM " + table.getTablename() + " t "
            + "WHERE ROWNUM <=" + lastPage + ") table_alias " +
                    " WHERE table_alias.rowno >=" + index);
    
            return  sb.toString();
        }
    
    
        /* * 数据量超过1W做分页循环插入 * */
    
        public static void putblocking(int sum, Connection con, Table table, BlockingQueue queue) throws Exception{ 
            String sql = "";
            //初始页数
            int index = 1;
            //最后一页
            int lastPage = PAGING;
            while (sum > 0){ 
                sql = getResultSet(index, lastPage, table);
                PreparedStatement ps = con.prepareStatement(sql);
                ResultSet rs = ps.executeQuery();
                //执行sql并放入队列中
                generateMysqlInsertSQLByTable(rs, table, queue);
                rs.close();
                ps.close();
    
                sum = sum -PAGING;
                index = lastPage + 1;
                if (sum > PAGING) { 
                    lastPage = lastPage + PAGING;
                } else { 
                    lastPage = lastPage + sum;
                }
    
            }
    
        }
    }

*  Consumer

/** * @author wh * @version 1.0 * @date 2019-12-16 10:08 */
    
    public class Consumer implements Runnable { 
        private BlockingQueue<String> blockingQueue;
        private Connection mysqlCon;
    
        public Consumer(BlockingQueue<String> blockingQueue  /*, Connection mysqlCon, boolean FLAG*/) { 
    
            try { 
    
                this.mysqlCon = Main.init(Main.MYSQL_DRIVER, Main.MYSQL_URL, Main.MYSQL_USERNAME, Main.MYSQL_PASSWORD);
    
    
                new Thread() { 
                    @Override
                    public void run() { 
                        while (true) { 
                            try { 
                                PreparedStatement preparedStatement = mysqlCon.prepareStatement("select 1");
                                ResultSet rs = preparedStatement.executeQuery();
                                System.out.println("执行一次 连接线程" + rs.next());
                                Thread.sleep(50 * 1000);
    
                                if (preparedStatement != null) { 
                                    preparedStatement.close();
                                }
                                if (rs != null) { 
                                    rs.close();
                                }
    
                            } catch (SQLException e) { 
                                e.printStackTrace();
                            } catch (InterruptedException e) { 
                                e.printStackTrace();
                            }
                        }
                    }
                }.start();
            } catch (Exception e) { 
                e.printStackTrace();
            }
            this.blockingQueue = blockingQueue;
        }
    
        @Override
        public void run() { 
            System.out.println("消费者启动");
            while (true) { 
                String mysqlSql = null;
                    try { 
                        mysqlSql = blockingQueue.take();
    
                        PreparedStatement preparedStatement = mysqlCon.prepareStatement(mysqlSql);
                        preparedStatement.executeUpdate();
                        preparedStatement.close();
                    } catch (Exception e) { 
                        e.printStackTrace();
                        System.out.println("消费者发生异常" );
    
                    }
    
            }
    
        }
    }

## 优化 ##

*  可选优化：

1.  使用线程池代替Thread.start
2.  将连接信息放入配置文件中
3.  运行注意分析GC日志，如果出现full gc 需要配置合理的JVM参数
4.  增多线程数

## 源码 ##

[github][]

[20200113154953304.png]: https://img-blog.csdnimg.cn/20200113154953304.png
[20200113155217466.png]: https://img-blog.csdnimg.cn/20200113155217466.png
[yugong]: https://github.com/alibaba/yugong
[20200113155347968.png]: https://img-blog.csdnimg.cn/20200113155347968.png
[20200113160234685.png]: https://img-blog.csdnimg.cn/20200113160234685.png
[github]: https://github.com/weihubeats/springcloud-shopping-parent/tree/master/springcloud-shopping-oracletomysql