java实现同步sqlserver中的表数据到mysql的方案

爱被打了一巴掌 2022-01-11 05:51 725阅读 0赞

# 场景 #

### 项目框架情况： ###

SpringMVC + MybatisPlus + msyql...

### 需求： ###

从sqlserver数据库中同步产品和产品类别两个表的数据。

# 思路 #

1、利用 java自带的java.sql.Connection 与sqlserver数据库建立连接。

2、通过Apache的dbutils工具类QueryRunner（ org.apache.commons.dbutils.QueryRunner）来获取sqlserver中的对应数据。

3、分析转化上述数据至我们需要的数据。

4、通过本地项目中的mysql数据源（或者步骤1中同样的方式），与msyql建立连接。

5、批量插入、更新mysql中的数据。

# 代码 #

1、数据源连接工具类DbConnUtil

package com.platform.syncdatabase.util;
    
    import com.platform.utils.ResourceUtil;
    
    import java.sql.Connection;
    import java.sql.DriverManager;
    
    
    public class DbConnUtil {
    
        private static String sqlserverDriverClassName = ResourceUtil.getConfigByName("hy.jindie.sqlserverDriverClassName");
        private static String sqlserverDbUrl = ResourceUtil.getConfigByName("hy.jindie.sqlserverDbUrl");
        private static String sqlserverDbUser = ResourceUtil.getConfigByName("hy.jindie.sqlserverDbUser");
        private static String sqlserverDbPwd = ResourceUtil.getConfigByName("hy.jindie.sqlserverDbPwd");
    
        /**
         * 从这里取数据
         *
         * @return
         */
        public static Connection createConnSrc() {
            Connection conn = null;
            try {
                Class.forName(sqlserverDriverClassName).newInstance();
                conn = DriverManager.getConnection(sqlserverDbUrl, sqlserverDbUser, sqlserverDbPwd);
            } catch (Exception e) {
                e.printStackTrace();
    
            }
            return conn;
        }
    
    
        /**
         * 存入的目标数据库
         *
         * @return
         */
        public static Connection createConnDist() {
            Connection conn = null;
            try {
                String driverClassName = "com.mysql.jdbc.Driver";
                String dbUrl = "jdbc:mysql://192.168.1.110:3306/haoyuan?useUnicode=true&characterEncoding=UTF-8";
                String dbUser = "root";
                String dbPwd = "1234567890";
                Class.forName(driverClassName).newInstance();
                conn = DriverManager.getConnection(dbUrl, dbUser, dbPwd);
            } catch (Exception e) {
                e.printStackTrace();
    
            }
    
            return conn;
        }
    
    
        public static void close(Connection conn) {
            try {
                conn.close();
            } catch (Exception e) {
                e.printStackTrace();
    
            }
        }
    
    }

2、Apache dbutils QueryRunner数据操作工具类的封装类DbBuilder

package com.platform.syncdatabase;
    
    
    import com.platform.syncdatabase.util.DbConnUtil;
    import org.apache.commons.dbutils.QueryRunner;
    import org.apache.commons.dbutils.handlers.BeanHandler;
    import org.apache.commons.dbutils.handlers.BeanListHandler;
    import org.apache.commons.dbutils.handlers.MapHandler;
    import org.apache.commons.dbutils.handlers.MapListHandler;
    
    import java.sql.Connection;
    import java.util.List;
    import java.util.Map;
    
    public class DbBuilder {
        Connection connection = null;
    
        public DbBuilder(Connection conn) {
            this.connection = conn;
        }
    
    
        public QueryRunner getQueryRunner() {
            return new QueryRunner();
        }
    
        /**
         * 根据传入的sql，查询记录，以Map形式返回第一行记录 注意：如果有多行记录，只会返回第一行，所以使用场景需要注意，可以使用根据主键来查询的场景
         *
         * @param sql
         * @param params
         * @return
         */
        public Map<String, Object> getFirstRowMap(String sql, Object... params) {
            try {
                QueryRunner runner = new QueryRunner();
                return runner.query(connection, sql, new MapHandler(), params);
            } catch (Exception ex) {
                ex.printStackTrace();
            }
            return null;
        }
    
    
        /**
         * 根据传入的sql查询数据，以List Map形式返回
         *
         * @param sql
         * @param params
         * @return
         */
        public List<Map<String, Object>> getListMap(String sql, Object... params) {
            try {
                QueryRunner runner = new QueryRunner();
                return runner.query(connection, sql, new MapListHandler(), params);
            } catch (Exception ex) {
                ex.printStackTrace();
            }
            return null;
        }
    
        /**
         * 根据sql和对象，查询结果并以对象形式返回
         *
         * @param sql
         * @param type
         * @return
         */
        public <T> T getBean(String sql, Class<T> type) {
            try {
                QueryRunner runner = getQueryRunner();
                return runner.query(sql, new BeanHandler<T>(type));
            } catch (Exception ex) {
                ex.printStackTrace();
            }
            return null;
        }
    
        /**
         * 根据sql和对象，查询结果并以对象形式返回
         *
         * @param sql
         * @param type
         * @return
         */
        public <T> T getBean(Connection connection, String sql, Class<T> type) {
            try {
                QueryRunner runner = new QueryRunner();
                return runner.query(connection, sql, new BeanHandler<T>(type));
            } catch (Exception ex) {
                ex.printStackTrace();
            }
            return null;
        }
    
    
        /**
         * 根据sql和对象，查询结果并以对象形式返回
         *
         * @param sql
         * @param type
         * @param params
         * @return
         */
        public <T> T getBean(Connection connection, String sql, Class<T> type, Object... params) {
            try {
                QueryRunner runner = new QueryRunner();
                return runner.query(connection, sql, new BeanHandler<T>(type), params);
            } catch (Exception ex) {
                ex.printStackTrace();
            }
            return null;
        }
    
    
        /**
         * 根据sql查询list对象
         *
         * @param sql
         * @param type
         * @return
         */
        public <T> List<T> getListBean(Connection connection, String sql, Class<T> type) {
            try {
                QueryRunner runner = new QueryRunner();
                return runner.query(connection, sql, new BeanListHandler<T>(type));
            } catch (Exception ex) {
                ex.printStackTrace();
            }
            return null;
        }
    
        /**
         * 根据sql查询list对象
         *
         * @param sql
         * @param type
         * @param params
         * @return
         */
        public <T> List<T> getListBean(String sql, Class<T> type, Object... params) {
            try {
                QueryRunner runner = getQueryRunner();
                return runner.query(sql, new BeanListHandler<T>(type), params);
            } catch (Exception ex) {
                ex.printStackTrace();
            }
            return null;
        }
    
        /**
         * 根据sql查询list对象
         *
         * @param sql
         * @param type
         * @param params
         * @return
         */
        public <T> List<T> getListBean(Connection connection, String sql, Class<T> type, Object... params) {
            try {
                QueryRunner runner = new QueryRunner();
                return runner.query(connection, sql, new BeanListHandler<T>(type), params);
            } catch (Exception ex) {
                ex.printStackTrace();
            }
            return null;
        }
    
        /**
         * 保存操作
         *
         * @param sql
         * @param params
         * @return
         */
        public int save(String sql, Object... params) {
            try {
                QueryRunner runner = getQueryRunner();
                return runner.update(connection, sql, params);
            } catch (Exception ex) {
                ex.printStackTrace();
            }
            return 0;
        }
    
        /**
         * 保存操作
         *
         * @param sql
         * @param params
         * @return
         */
        public int save(Connection connection, String sql, Object... params) {
            try {
                QueryRunner runner = new QueryRunner();
                return runner.update(connection, sql, params);
            } catch (Exception ex) {
                ex.printStackTrace();
            }
            return 0;
        }
    
        /**
         * 更新操作
         *
         * @param sql
         * @param params
         * @return
         */
        public int update(String sql, Object... params) {
            try {
                QueryRunner runner = getQueryRunner();
                return runner.update(sql, params);
            } catch (Exception ex) {
                ex.printStackTrace();
            }
            return 0;
        }
    
        /**
         * 更新操作
         *
         * @param sql
         * @param params
         * @return
         */
        public int update(Connection connection, String sql, Object... params) {
            try {
                QueryRunner runner = new QueryRunner();
                return runner.update(connection, sql, params);
            } catch (Exception ex) {
                ex.printStackTrace();
            }
            return 0;
        }
    
        /**
         * 删除操作
         *
         * @param sql
         * @param params
         * @return
         */
        public int delete(String sql, Object... params) {
            try {
                QueryRunner runner = getQueryRunner();
                return runner.update(sql, params);
            } catch (Exception ex) {
                ex.printStackTrace();
            }
            return 0;
        }
    
        /**
         * 删除操作
         *
         * @param sql
         * @param params
         * @return
         */
        public int delete(Connection connection, String sql, Object... params) {
            try {
                QueryRunner runner = new QueryRunner();
                return runner.update(connection, sql, params);
            } catch (Exception ex) {
                ex.printStackTrace();
            }
            return 0;
        }
    
        /**
         * 批量操作，包括批量保存、修改、删、	 *
         *
         * @param sql
         * @param params
         * @return
         */
        public int[] batch(String sql, Object[][] params) {
            try {
                QueryRunner runner = getQueryRunner();
                return runner.batch(sql, params);
            } catch (Exception ex) {
                ex.printStackTrace();
            }
            return null;
        }
    
        /**
         * 批量操作，包括批量保存、修改、删处	 *
         *
         * @param sql
         * @param params
         * @return
         */
        public int[] batch(Connection connection, String sql, Object[][] params) {
            try {
                QueryRunner runner = new QueryRunner();
                return runner.batch(connection, sql, params);
            } catch (Exception ex) {
                ex.printStackTrace();
            }
            return null;
        }
    
        /**
         * �?��事务
         */
        public void beginTransaction(Connection conn) {
            try {
                conn.setAutoCommit(false);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
    
        /**
         * 回滚事务
         */
        public void rollback(Connection conn) {
            try {
                conn.rollback();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
    
        /**
         * 提交事务
         */
        public void commit(Connection conn) {
            try {
                conn.commit();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
    
        public static void main(String[] args) {
            try {
    
                Connection conn = DbConnUtil.createConnDist();
                DbBuilder b = new DbBuilder(conn);
                System.out.println(b.save("insert into demo_test(id,kitchen) values(?,?)", "11", "11"));
                DbConnUtil.close(conn);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

3、数据同步类DoDataSycn，以下同步中对mysql中已存在的数据进行更新时可以先与sqlserver中的数据对比版本变化，只更新有变化的数据即可。此处由于sqlserver数据库的数据版本信息不可用所以取消了对比，而是全量更新。

package com.platform.syncdatabase;
    
    import com.platform.entity.CategoryEntity;
    import com.platform.entity.GoodsEntity;
    import com.platform.service.CategoryService;
    import com.platform.service.GoodsService;
    import com.platform.syncdatabase.util.DbConnUtil;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.commons.lang.StringUtils;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import java.math.BigDecimal;
    import java.sql.Connection;
    import java.util.*;
    
    @Slf4j
    public class DoDataSycn {
    
        @Autowired
        CategoryService categoryService;
    
        @Autowired
        GoodsService goodsService;
    
        /**
         * 更新品类
         */
        public void updateCategory() {
            //打开链接
            Connection connSqlserver = DbConnUtil.createConnSrc();
            DbBuilder sqlserver = new DbBuilder(connSqlserver);
            try {
                //DO
                //DO 1、遍历mysql数据。添加map id为key
                //DO 2、遍历sqlserver数据获取新增数据添加插入map，获取修改过的数据添加update map
                //DO 3、实现批量更新功能
                String sql = "Select * from t_Item WHERE FItemClassID = 4";
                List<Map<String, Object>> remoteList = sqlserver.getListMap(sql, null);//获取远程金蝶数据
                if (remoteList != null && !remoteList.isEmpty()) {
                    List<CategoryEntity> insertList = new ArrayList<>();
                    List<CategoryEntity> updateList = new ArrayList<>();
                    //查询mysql现有品类
                    List<CategoryEntity> mysqlCategoryList = categoryService.queryList(new HashMap<>());
                    if (mysqlCategoryList != null && !mysqlCategoryList.isEmpty()) {//已存在数据时，对比更新
                        //1、现有数据分析整理
                        Map<String, CategoryEntity> mysqlCategoryMap = new HashMap<>();
                        for (CategoryEntity category : mysqlCategoryList) {
                            mysqlCategoryMap.put(category.getId() + "", category);
                        }
                        //获取现有id
                        Set<String> ids = mysqlCategoryMap.keySet();
                        log.info("ids.size() " + ids.size());
    
                        //2、开始对比数据
                        //遍历金蝶数据
                        for (Map item : remoteList) {
                            try {
                                String id = item.get("FItemID") + "";
                                if (ids.contains(id)) {//已存在的品类 更新
                                    updateList.add(copyCategoryInfo(item, mysqlCategoryMap.get(id)));
                                } else {//新增的品类
                                    insertList.add(copyCategoryInfo(item, new CategoryEntity()));
                                }
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }
    
                    } else {//初始化全量插入
                        for (Map item : remoteList) {
                            try {
                                insertList.add(copyCategoryInfo(item, new CategoryEntity()));
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }
                    }
                    try {
                        if (!insertList.isEmpty()) {
                            log.info(("saveBatch category " + insertList.size()));
                            categoryService.saveBatch(insertList);
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
    
                    try {
                        if (!updateList.isEmpty()) {
                            log.info(("batchUpdate category " + updateList.size()));
                            categoryService.batchUpdate(updateList);
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
    
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                //关闭
                DbConnUtil.close(connSqlserver);
            }
    
        }
    
        /**
         * 更新品类信息
         *
         * @param item
         * @param categoryEntity
         * @return
         */
        private CategoryEntity copyCategoryInfo(Map item, CategoryEntity categoryEntity) {
            categoryEntity.setId(Integer.parseInt(item.get("FItemID") + ""));
            categoryEntity.setName(item.get("FName") + "");
            categoryEntity.setParentId(Integer.parseInt(item.get("FParentID") + ""));
            categoryEntity.setSortOrder(Integer.parseInt(item.get("FShortNumber") + ""));
            String fDeleted = item.get("FDeleted") + "";
            if (StringUtils.isNotBlank(fDeleted) && "1".equals(fDeleted)) {
                categoryEntity.setIsShow(0);
            } else {
                categoryEntity.setIsShow(1);
            }
            categoryEntity.setLevel(item.get("FLevel") + "");
            categoryEntity.setFlevel(Integer.parseInt(item.get("FLevel") + ""));
            categoryEntity.setFullnumber(item.get("FFullNumber") + "");
            categoryEntity.setFullname(item.get("FFullName") + "");
            categoryEntity.setUuid(item.get("UUID") + "");
            //拼接ModifyTime
            byte[] fModifyTimes = (byte[]) item.get("FModifyTime");
            StringBuilder modifyTime = new StringBuilder();
            for (byte b : fModifyTimes) {
                modifyTime.append(b);
            }
            categoryEntity.setModifyTime(modifyTime.toString());//添加更新版本信息
            categoryEntity.setIsGood((boolean) item.get("FDetail") ? "1" : "0");
            return categoryEntity;
        }
    
    
        /**
         * 更新产品
         */
        public void updateGoods() {
            //打开链接
            Connection connSqlserver = DbConnUtil.createConnSrc();
            DbBuilder sqlserver = new DbBuilder(connSqlserver);
            try {
                //DO
                //DO 1、遍历mysql数据。添加map id为key
                //DO 2、遍历sqlserver数据获取新增数据添加插入map，获取修改过的数据添加update map
                //DO 3、实现批量更新功能
                String sql = "SELECT" +
                        " i.FItemID," +
                        " c.FName," +
                        " i.FParentID," +
                        " i.FShortNumber," +
                        " i.FDeleted," +
                        " i.FLevel," +
                        " i.FFullNumber," +
                        " i.FFullName," +
                        " i.UUID," +
                        " i.FModifyTime," +
                        " c.FOrderPrice," +
                        " c.FOrderPrice," +
                        " c.FOrderPrice," +
                        " c.FModel " +
                        "FROM " +
                        " t_item i" +
                        " LEFT JOIN t_icitemcore c ON i.FItemID = c.FItemID " +
                        "WHERE " +
                        " i.FItemClassID = 4 " +
                        " AND i.FDetail = 1 ;";
    //                    " AND i.FDetail = 1 "+
    //                    " AND i.FItemID = 291;";
                List<Map<String, Object>> remoteList = sqlserver.getListMap(sql, null);//获取远程金蝶数据
                if (remoteList != null && !remoteList.isEmpty()) {
                    List<GoodsEntity> insertList = new ArrayList<>();
                    List<GoodsEntity> updateList = new ArrayList<>();
                    //查询mysql现有产品
                    List<GoodsEntity> mysqlGoodsList = goodsService.queryAllGoods();
                    if (mysqlGoodsList != null && !mysqlGoodsList.isEmpty()) {//已存在数据时，对比更新
                        //1、现有数据分析整理
                        Map<String, GoodsEntity> mysqlGoodsMap = new HashMap<>();
                        for (GoodsEntity goods : mysqlGoodsList) {
                            mysqlGoodsMap.put(goods.getId() + "", goods);
                        }
                        //获取现有id
                        Set<String> ids = mysqlGoodsMap.keySet();
    
                        //2、开始对比数据
                        //遍历金蝶数据
                        for (Map item : remoteList) {
                            try {
                                String id = item.get("FItemID") + "";
                                if (ids.contains(id)) {//已存在的品类
                                    //更新
                                    updateList.add(copyGoodsInfo(item, mysqlGoodsMap.get(id)));
                                } else {//新增的品类
                                    insertList.add(copyGoodsInfo(item, new GoodsEntity()));
                                }
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }
    
                    } else {//初始化全量插入
                        for (Map item : remoteList) {
                            try {
                                insertList.add(copyGoodsInfo(item, new GoodsEntity()));
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }
                    }
                    try {
                        if (!insertList.isEmpty()) {
                            log.info(("saveBatch goods " + insertList.size()));
                            goodsService.saveBatch(insertList);
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
    
                    try {
                        if (!updateList.isEmpty()) {
                            log.info(("batchUpdate goods " + updateList.size()));
                            goodsService.batchUpdate(updateList);
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
    
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                //关闭
                DbConnUtil.close(connSqlserver);
            }
    
        }
    
    
        /**
         * 更新产品信息
         *
         * @param item
         * @param goodsEntity
         * @return
         */
        private GoodsEntity copyGoodsInfo(Map item, GoodsEntity goodsEntity) {
            goodsEntity.setId(Integer.parseInt(item.get("FItemID") + ""));
            goodsEntity.setName(item.get("FName") + "");
            goodsEntity.setCategoryId(Integer.parseInt(item.get("FParentID") + ""));
            goodsEntity.setSortOrder(Integer.parseInt(item.get("FShortNumber") + ""));
            String fDeleted = item.get("FDeleted") + "";
            if (StringUtils.isNotBlank(fDeleted) && "1".equals(fDeleted)) {
                goodsEntity.setIsDelete(0);
            } else {
                goodsEntity.setIsDelete(1);
            }
            goodsEntity.setFlevel(Integer.parseInt(item.get("FLevel") + ""));
            goodsEntity.setFullnumber(item.get("FFullNumber") + "");
            goodsEntity.setFullname(item.get("FFullName") + "");
            goodsEntity.setUuid(item.get("UUID") + "");
            //拼接ModifyTime
            byte[] fModifyTimes = (byte[]) item.get("FModifyTime");
            StringBuilder modifyTime = new StringBuilder();
            for (byte b : fModifyTimes) {
                modifyTime.append(b);
            }
            goodsEntity.setModifyTime(modifyTime.toString());//添加更新版本信息
            goodsEntity.setCounterPrice(new BigDecimal(item.get("FOrderPrice") + ""));
            goodsEntity.setRetailPrice(new BigDecimal(item.get("FOrderPrice") + ""));
            goodsEntity.setUnitPrice(new BigDecimal(item.get("FOrderPrice") + ""));
            goodsEntity.setModel(item.get("FModel") + "");
            return goodsEntity;
        }
    
    
    }

4、项目properties配置文件中的sqlserver连接信心配置

#sqlserver连接信息
    hy.jindie.sqlserverDriverClassName = net.sourceforge.jtds.jdbc.Driver
    hy.jindie.sqlserverDbUrl =  jdbc:jtds:sqlserver://192.168.1.111:1433;DatabaseName=haoyuan
    hy.jindie.sqlserverDbUser = sa
    hy.jindie.sqlserverDbPwd =  123213214123

在此鸣谢开源作者：[在水一方发盐人][Link 1] 的开源代码。[https://gitee.com/lostad/sqlserver\_to\_mysql/tree/master][https_gitee.com_lostad_sqlserver_to_mysql_tree_master]

[Link 1]: https://gitee.com/lostad
[https_gitee.com_lostad_sqlserver_to_mysql_tree_master]: https://gitee.com/lostad/sqlserver_to_mysql/tree/master