一、資料同步簡介

1、場景描述

如果經常接觸資料開發，會有這樣一個場景，服務A提供一個資料源，假設稱為動态資料源A，需要讀取該資料源下的資料；服務B提供一個資料源，假設稱為動态資料源B，需要寫入資料到該資料源。這個場景通常描述為資料同步，或者資料搬運。

2、基本流程

資料源管理 | 動态權限校驗，表結構和資料遷移流程一、資料同步簡介二、基礎工具封裝三、業務化流程四、最後總結五、源代碼位址

基于上述流程圖，整體步驟如下：

測試多個資料源是否連接配接成功，并動态管理;
判斷資料源提供的賬号是否有操作權限，例如讀寫;
讀取資料源A的表結構，在資料源B建立表;
資料讀取或者分頁讀取，寫入資料源B中;
在不知道表結構情況下，還需要讀取表結構，生成SQL;

3、JDBC基礎API

Statement

Java中JDBC下執行資料庫操作的一個重要接口，在已經建立資料庫連接配接的基礎上，向資料庫發送要執行的SQL語句。

PreparedStatement

繼承Statement接口，且實作SQL預編譯，可以提高批量處理效率。常應用于批量資料寫入場景。

ResultSet

存儲JDBC查詢結果集的對象，ResultSet接口提供從目前行檢索列值的方法。

二、基礎工具封裝

1、資料源管理

提供一個資料源管理的Factory，目前場景下主要管理一個讀庫即資料源A，和一個寫庫即資料源B，資料源連接配接驗證通過，放入容器中。

@Component
public class ConnectionFactory {

    private volatile Map<String, Connection> connectionMap = new HashMap<>();

    @Resource
    private JdbcConfig jdbcConfig ;

    @PostConstruct
    public void init (){
        ConnectionEntity read = new ConnectionEntity(
        "MySql","jdbc:mysql://localhost:3306/data_read","user01","123");
        if (jdbcConfig.getConnection(read) != null){
            connectionMap.put(JdbcConstant.READ,jdbcConfig.getConnection(read));
        }
        ConnectionEntity write = new ConnectionEntity(
        "MySql","jdbc:mysql://localhost:3306/data_write","user01","123");
        if (jdbcConfig.getConnection(write) != null){
            connectionMap.put(JdbcConstant.WRITE,jdbcConfig.getConnection(write));
        }
    }

    public Connection getByKey (final String key){
        return connectionMap.get(key) ;
    }
}

2、動态SQL拼接

基礎SQL管理

主要提供SQL的基礎模闆，例如全表查，分頁查，表結構查詢。

public class BaseSql {
    public static String READ_SQL = "SELECT * FROM %s LIMIT 1";
    public static String WRITE_SQL = "INSERT INTO %s (SELECT * FROM %s WHERE 1=0)" ;
    public static String CREATE_SQL = "SHOW CREATE TABLE %s" ;
    public static String SELECT_SQL = "SELECT * FROM %s" ;
    public static String COUNT_SQL = "SELECT COUNT(1) countNum FROM %s" ;
    public static String PAGE_SQL = "SELECT * FROM %s LIMIT %s,%s" ;
    public static String STRUCT_SQL (){
        StringBuffer sql = new StringBuffer() ;
        sql.append(" SELECT                     ");
        sql.append("     COLUMN_NAME,           ");
        sql.append("     IS_NULLABLE,           ");
        sql.append("     COLUMN_TYPE,           ");
        sql.append("     COLUMN_KEY,            ");
        sql.append("     COLUMN_COMMENT         ");
        sql.append(" FROM                       ");
        sql.append(" information_schema.COLUMNS ");
        sql.append(" WHERE                      ");
        sql.append(" table_schema = '%s'        ");
        sql.append(" AND table_name = '%s'      ");
        return String.valueOf(sql) ;
    }
}

SQL參數拼接

根據SQL模闆中缺失的參數，進行動态補全，生成完成SQL語句。

public class BuildSql {
    /**
     * 讀權限SQL
     */
    public static String buildReadSql(String table) {
        String readSql = null ;
        if (StringUtils.isNotEmpty(table)){
            readSql = String.format(BaseSql.READ_SQL, table);
        }
        return readSql;
    }
    /**
     * 讀權限SQL
     */
    public static String buildWriteSql(String table){
        String writeSql = null ;
        if (StringUtils.isNotEmpty(table)){
            writeSql = String.format(BaseSql.WRITE_SQL, table,table);
        }
        return writeSql ;
    }
    /**
     * 表建立SQL
     */
    public static String buildStructSql (String table){
        String structSql = null ;
        if (StringUtils.isNotEmpty(table)){
            structSql = String.format(BaseSql.CREATE_SQL, table);
        }
        return structSql ;
    }
    /**
     * 表結構SQL
     */
    public static String buildTableSql (String schema,String table){
        String structSql = null ;
        if (StringUtils.isNotEmpty(table)){
            structSql = String.format(BaseSql.STRUCT_SQL(), schema,table);
        }
        return structSql ;
    }
    /**
     * 全表查詢SQL
     */
    public static String buildSelectSql (String table){
        String selectSql = null ;
        if (StringUtils.isNotEmpty(table)){
            selectSql = String.format(BaseSql.SELECT_SQL,table);
        }
        return selectSql ;
    }
    /**
     * 總數查詢SQL
     */
    public static String buildCountSql (String table){
        String countSql = null ;
        if (StringUtils.isNotEmpty(table)){
            countSql = String.format(BaseSql.COUNT_SQL,table);
        }
        return countSql ;
    }
    /**
     * 分頁查詢SQL
     */
    public static String buildPageSql (String table,int offset,int size){
        String pageSql = null ;
        if (StringUtils.isNotEmpty(table)){
            pageSql = String.format(BaseSql.PAGE_SQL,table,offset,size);
        }
        return pageSql ;
    }
}

三、業務化流程

1、基礎鑒權

讀庫嘗試一次單條資料讀取，寫庫嘗試一次不成立條件的寫入，如果沒有權限，會抛出相應異常。

@RestController
public class CheckController {
    @Resource
    private ConnectionFactory connectionFactory ;
    // MySQLSyntaxErrorException: SELECT command denied to user
    @GetMapping("/checkRead")
    public String checkRead (){
        try {
            String sql = BuildSql.buildReadSql("rw_read") ;
            ExecuteSqlUtil.query(connectionFactory.getByKey(JdbcConstant.READ),sql) ;
            return "success" ;
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return "fail" ;
    }
    // MySQLSyntaxErrorException: INSERT command denied to user
    @GetMapping("/checkWrite")
    public String checkWrite (){
        try {
            String sql = BuildSql.buildWriteSql("rw_read") ;
            ExecuteSqlUtil.update(connectionFactory.getByKey(JdbcConstant.WRITE),sql) ;
            return "success" ;
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return "fail" ;
    }
}

2、同步表結構

這裡執行最簡單操作，把讀庫表建立語句查詢出來，丢到寫庫中執行。

@RestController
public class StructController {
    @Resource
    private ConnectionFactory connectionFactory ;
    @GetMapping("/syncStruct")
    public String syncStruct (){
        try {
            String sql = BuildSql.buildStructSql("rw_read") ;
            ResultSet resultSet = ExecuteSqlUtil.query(connectionFactory.getByKey(JdbcConstant.READ),sql) ;
            String createTableSql = null ;
            while (resultSet.next()){
                createTableSql = resultSet.getString("Create Table") ;
            }
            if (StringUtils.isNotEmpty(createTableSql)){
                ExecuteSqlUtil.update(connectionFactory.getByKey(JdbcConstant.WRITE),createTableSql) ;
            }
            return "success" ;
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return "fail" ;
    }
}

3、同步表資料

讀庫的表資料讀取，批量放入寫庫中。這裡特别說一個方法：statement.setObject();在不知道參數個數和類型時，自動适配資料類型。

@RestController
public class DataSyncController {
    @Resource
    private ConnectionFactory connectionFactory ;
    @GetMapping("/dataSync")
    public List<RwReadEntity> dataSync (){
        List<RwReadEntity> rwReadEntities = new ArrayList<>() ;
        try {
            Connection readConnection = connectionFactory.getByKey(JdbcConstant.READ) ;
            String sql = BuildSql.buildSelectSql("rw_read") ;
            ResultSet resultSet = ExecuteSqlUtil.query(readConnection,sql) ;
            while (resultSet.next()){
                RwReadEntity rwReadEntity = new RwReadEntity() ;
                rwReadEntity.setId(resultSet.getInt("id"));
                rwReadEntity.setSign(resultSet.getString("sign"));
                rwReadEntities.add(rwReadEntity) ;
            }
            if (rwReadEntities.size() > 0){
                Connection writeConnection = connectionFactory.getByKey(JdbcConstant.WRITE) ;
                writeConnection.setAutoCommit(false);
                PreparedStatement statement = writeConnection.prepareStatement("INSERT INTO rw_read VALUES(?,?)");
                // 基于動态擷取列，和statement.setObject();自動适配資料類型
                for (int i = 0 ; i < rwReadEntities.size() ; i++){
                    RwReadEntity rwReadEntity = rwReadEntities.get(i) ;
                    statement.setInt(1,rwReadEntity.getId()) ;
                    statement.setString(2,rwReadEntity.getSign()) ;
                    statement.addBatch();
                    if (i>0 && i%2==0){
                        statement.executeBatch() ;
                    }
                }
                // 處理最後一批資料
                statement.executeBatch();
                writeConnection.commit();
            }
            return rwReadEntities ;
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return null ;
    }
}

4、分頁查詢

提供一個分頁查詢工具，在資料量大的情況下不能一次性讀取大量的資料，避免資源占用過高。

public class PageUtilEntity {
    /**
     * 分頁生成方法
     */
    public static PageHelperEntity<Object> pageResult (int total, int pageSize,int currentPage, List dataList){
        PageHelperEntity<Object> pageBean = new PageHelperEntity<Object>();
        // 總頁數
        int totalPage = PageHelperEntity.countTotalPage(pageSize,total) ;
        // 分頁清單
        List<Integer> pageList = PageHelperEntity.pageList(currentPage,pageSize,total) ;
        // 上一頁
        int prevPage = 0 ;
        if (currentPage==1){
            prevPage = currentPage ;
        } else if (currentPage>1&&currentPage<=totalPage){
            prevPage = currentPage -1 ;
        }
        // 下一頁
        int nextPage =0 ;
        if (totalPage==1){
            nextPage = currentPage ;
        } else if (currentPage<=totalPage-1){
            nextPage = currentPage+1 ;
        }
        pageBean.setDataList(dataList);
        pageBean.setTotal(total);
        pageBean.setPageSize(pageSize);
        pageBean.setCurrentPage(currentPage);
        pageBean.setTotalPage(totalPage);
        pageBean.setPageList(pageList);
        pageBean.setPrevPage(prevPage);
        pageBean.setNextPage(nextPage);
        pageBean.initjudge();
        return  pageBean ;
    }
}

四、最後總結

很多複雜度偏高的業務，越是需要借助基礎API解決，因為複雜度高，不容易抽象化統一封裝，如果資料同步這塊業務，可以适配多種資料庫，完全可以獨立封裝為中間件，開源項目中關于多方資料同步或計算的中間件也有好多，可以自行了解下，增長眼界開闊思路。

五、源代碼位址

GitHub·位址
https://github.com/cicadasmile/data-manage-parent
GitEE·位址
https://gitee.com/cicadasmile/data-manage-parent

資料源管理 | 動态權限校驗，表結構和資料遷移流程一、資料同步簡介二、基礎工具封裝三、業務化流程四、最後總結五、源代碼位址

一、資料同步簡介

1、場景描述

2、基本流程

3、JDBC基礎API

二、基礎工具封裝

1、資料源管理

2、動态SQL拼接

三、業務化流程

1、基礎鑒權

2、同步表結構

3、同步表資料

4、分頁查詢

四、最後總結

五、源代碼位址

繼續閱讀

neo4j之cypher使用文檔

GitHub連夜封殺！這份阿裡 10W 字内部 Java 字面試手冊到底有多強？

spark/scala關于【資源檔案】加載方法概述外部檔案加載方案測試資源檔案打包入jar包中小結

NOSQL安全攻擊

mybatis_入門程式Mybatis入門

mysql使用source指令導入.sql檔案

AOP程式設計_Android優雅權限架構(1)概念基礎，2021金三銀四前言正文大綱正文

登入plsql 報錯 the account is locked --使用者被鎖

sqlServer根據經緯查距離

Effective Java 8:通用程式設計

SequoiaDB巨杉資料庫C++驅動概述

OOM三種類型

工廠模式-三種類型

【遞歸】高效率求2的n次幂

win10本地scala和spark安裝安裝scala安裝spark

scala (3) Function 和 Method