数据同步代码（Java实现）

ゞ浴缸里的玫瑰 2022-05-28 02:52 320阅读 0赞

业务场景：  
从15.5数据库把需要同步的数据表同步到174数据表中，其中174中有些字段比较短，有些字段不存在，有些表不存在，重复主键等问题。  
字段大小不一致：执行insert操作时捕获异常，忽略错误。  
174表中字段不一致:判断表中是否有相应的字段。  
表是否存在：判断表是否存在。  
重复主键：原本通过delete，删除了以前的数据。后来使用了try catch捕获异常，忽略错误。这里有些可以改进的地方，有更新的情况没有考虑。  
System.out.println()严重影响了效率，尽量避免使用。  
其实以上的错误都可以通过在insert时捕获异常来处理，代码的处理逻辑与上面的描述一致。

主函数

package dataChange;
    
    import java.sql.Connection;
    import java.sql.SQLException;
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    
    public class ChangeMain { 
        public static void main(String[] args)
        {
            GetMessage gm = new GetMessage();
            try {
            //从源数据库中获取数据
                Map<String, List<HashMap<String, Object>>> map = gm.selectData();
            //插入到目标数据库中
                gm.insertNewDataBase(map);
            } catch (SQLException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
    
        }
    
    
    
    
    }

获取数据和插入数据：

package dataChange;
    
    import java.sql.Connection;
    import java.sql.DriverManager;
    import java.sql.ResultSet;
    import java.sql.ResultSetMetaData;
    import java.sql.SQLException;
    import java.sql.Statement;
    import java.util.ArrayList;
    import java.util.HashMap;
    import java.util.Hashtable;
    import java.util.List;
    import java.util.Map;
    import java.util.Map.Entry;
    
    public class GetMessage {
        //从15.5上获取数据
    
        public Map<String, List<HashMap<String, Object>>> selectData() throws SQLException {
                Connection con = getConnection("","","","");
                Statement stmt  = con.createStatement();
                //dx_dbdata_bxxpz表中存放着需要同步的数据表
                ResultSet rs =stmt.executeQuery("select * from dx_dbdata_bxxpz");
    
                List<String> li = new ArrayList<>();
    
    
                while (rs.next())
                {
                    //第二列数据为数据表名称
                    li.add(rs.getString(2));
                }
    
                rs.close();
                stmt.close();
    
    
            /* while (rs.next()) { Hashtable<String, Object> row = new Hashtable<String, Object>(); for (int i = 1; i <= rsm.getColumnCount(); i++) { // 打印列名 System.out.println(rsm.getColumnName(i)); row.put(rsm.getColumnName(i), rs.getObject(i)); } rows.add(row); }*/
                //这个map 键为表的名称，值为表中所有数据的list集合，list集合包含map，其中键为列名称，值为列对应的值
                Map<String,List<HashMap<String, Object>>> totalMap = new HashMap<>();
                for (int i = 0;i < li.size();i++)
                {
    
                    Statement stmt1  = con.createStatement();
                    String tableName = li.get(i);
                    String sql =  "select * from "+tableName;
                    ResultSet rs1 =stmt1.executeQuery(sql);
                    ResultSetMetaData rsm = rs1.getMetaData();
                    //这里List，便于保存
                    List<HashMap<String, Object>> rows = new ArrayList<HashMap<String, Object>>();
                    while (rs1.next())
                    {
                        HashMap<String, Object> row = new HashMap<String, Object>();
                        for (int j = 1; j <= rsm.getColumnCount(); j++) {
    
    
                                row.put(rsm.getColumnName(j), rs1.getObject(j));
    
                        }
                        rows.add(row);
    
                    }
    
    
                    totalMap.put(tableName, rows);
                // System.out.println(totalMap);
                    rs1.close();
                    stmt1.close();
    
    
                }
                con.close();
    
    
                return totalMap;
    
        }
        //实现数据库的连接
        private static Connection getConnection(String url,String  user,String password,String shema) {
            String driver = "com.mysql.jdbc.Driver";
            String url1 = "jdbc:mysql://"+url+":3306/"+shema;
            Connection con = null;
            try {
                Class.forName(driver);
                con = DriverManager.getConnection(url1, user, password);
            } catch (ClassNotFoundException | SQLException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            return con;
    
    
    
        }
        //插入到174的环境中
        public void insertNewDataBase(Map<String, List<HashMap<String, Object>>> map)  {
            Connection con =getConnection("","","","");
            for (Entry<String,List<HashMap<String, Object>>> elh:map.entrySet())
            {
    
    
    
                String key = elh.getKey();
    
                List<HashMap<String, Object>> lhso = elh.getValue();
                //先执行删除操作
    
                try {
                    if (exitsTable(key,con)){
    
                            //拼接insert into的列明1，列明2.......和value1....。这里使用StringBuilder，StringBuilder貌似是线程不安全的。方便不容易出错
    
                            StringBuilder sb = new StringBuilder();
                            StringBuilder sb1 = new StringBuilder();
                            for (int i = 0;i < lhso.size();i++)
                            {
    
    
                                HashMap<String, Object> mapColumn = lhso.get(i);
                                for (Entry<String, Object>  eso:mapColumn.entrySet()){
    
    
                                    try {
                                        if (columAndJudge(key,eso.getKey()))
                                        {
                                            sb.append(eso.getKey());
                                            sb.append(",");
                                            if (eso.getValue() != null && (eso.getValue().getClass().getName().equals("java.lang.String")))
                                            {
    
                                                    String result = "'"+eso.getValue()+"'";
                                                    sb1.append(result);
    
    
    
                                            }else
                                            {
    
                                                    sb1.append(eso.getValue()); 
                                            }
    
                                            sb1.append(",");
    
                                        }
                                    } catch (SQLException e) {
                                        // TODO Auto-generated catch block
                                        e.printStackTrace();
                                    }
    
    
    
    
    
                                }
                                sb.deleteCharAt(sb.length()-1);  
                                sb1.deleteCharAt(sb1.length()-1);
                                String column = sb.toString();
                                String value = sb1.toString();
                                Statement stmtInsert = null;
                                try {
                                    stmtInsert = con.createStatement();
                                    String sql = "insert into "+key+"("+column+")"+"values ("+value+")";
                                    stmtInsert.executeUpdate(sql);
                                } catch (SQLException e) {
                                //这里一定要删除，否组会出错
                                    sb =  sb.delete( 0, sb.length() );
                                sb1 = sb1.delete( 0, sb1.length() );
                                    System.out.println("忽略错误"+key);
                                    continue;
                                } finally{
                                    try {
                                        stmtInsert.close();
                                    } catch (SQLException e) {
                                        // TODO Auto-generated catch block
                                        e.printStackTrace();
                                    }
                                }
                                //INSERT INTO table_name (列1, 列2,...) VALUES (值1, 值2,....)
    
            //这里使用delete清空StringBuilder，网上搜了一下，貌似是清空中效率最高的 
    
                                sb =  sb.delete( 0, sb.length() );
                                sb1 = sb1.delete( 0, sb1.length() );
    
    
    
    
    
                            }
    
    
    
    
    
    
                        }
                } catch (SQLException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
    
            }
            try {
                con.close();
            } catch (SQLException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            System.out.println("同步完成");
    
        }
        //判断是否存在特定的数据表
        private boolean exitsTable(String key, Connection con) throws SQLException {
    
            ResultSet rs  = con.getMetaData().getTables(null, null,  key, null );
            if (rs.next())
            {
                return true;
            }else
            {
                return false;
            }
        }
        /** * 判断执行的表格是否特定的列名 * @param key 表名称 * @param key2 列名称 * @return * @throws SQLException */
        private boolean columAndJudge(String key, String key2) throws SQLException {
    
            Connection con =getConnection("","","","");
            Statement stmt = con.createStatement();
            String sql = "select * from "+key;
            ResultSet rs = stmt.executeQuery(sql);
            ResultSetMetaData  rmd = rs.getMetaData();
    
    
    
    
    
                for (int i = 1; i <= rmd.getColumnCount(); i++) {
                    // 打印列名
                        if (key2.equals(rmd.getColumnName(i)))
                        {
                            stmt.close();
                            con.close();
                            return true;
    
                        }
                    }
                stmt.close();
                con.close();
                return false;
    
    
    
    
        }
    
    
    
    }

以后可以研究下线程池和数据库连接池，这样性能不太好。有很大改进空间，希望不吝指教。  
考虑 过使用批量插入的方法，但实际上相应的列对应不上。  
批量插入代码实例：

import com.mysql.jdbc.Connection;
    
    import java.io.IOException;
    import java.sql.DriverManager;
    import java.sql.PreparedStatement;
    import java.sql.SQLException;
    
    public class MysqlBatchUtil { 
    
        private String sql = "INSERT INTO intest (username, password, create_time) VALUES (?, ?, now())";
        //MySql 的批量操作，要加rewriteBatchedStatements参数
        private String connectStr = "jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=utf-8&useServerPrepStmts=false&rewriteBatchedStatements=true";
        private String username = "root";
        private String password = "123";
    
        private void doStore() throws ClassNotFoundException, SQLException, IOException {
            Class.forName("com.mysql.jdbc.Driver");
            Connection conn = (Connection) DriverManager.getConnection(connectStr, username, password);
            conn.setAutoCommit(false);
            int count = 0;
            PreparedStatement psts = conn.prepareStatement(sql);
            long start = System.currentTimeMillis();
            for (int i = 1; i <= 5000000; i++) {
                psts.setString(1, "user" + i);
                psts.setString(2, "pass" + i);
                psts.addBatch();   // 加入批量处理
                count++;
            }
            psts.executeBatch(); // 执行批量处理
            conn.commit(); // 提交
            long end = System.currentTimeMillis();
            System.out.println("数量=" + count);
            System.out.println("运行时间=" + (end - start));
            conn.close();
        }
    
        public static void main(String[] args) {
            try {
                new MysqlBatchUtil().doStore();
            } catch (ClassNotFoundException | SQLException | IOException e) {
                e.printStackTrace();
            }
        }
    }