本文源碼：GitHub·點(diǎn)這里 || GitEE·點(diǎn)這里

一、數(shù)據(jù)同步簡(jiǎn)介

1、場(chǎng)景描述

如果經(jīng)常接觸數(shù)據(jù)開(kāi)發(fā)，會(huì)有這樣一個(gè)場(chǎng)景，服務(wù)A提供一個(gè)數(shù)據(jù)源，假設(shè)稱(chēng)為動(dòng)態(tài)數(shù)據(jù)源A，需要讀取該數(shù)據(jù)源下的數(shù)據(jù)；服務(wù)B提供一個(gè)數(shù)據(jù)源，假設(shè)稱(chēng)為動(dòng)態(tài)數(shù)據(jù)源B，需要寫(xiě)入數(shù)據(jù)到該數(shù)據(jù)源。這個(gè)場(chǎng)景通常描述為數(shù)據(jù)同步，或者數(shù)據(jù)搬運(yùn)。

2、基本流程

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

• 測(cè)試多個(gè)數(shù)據(jù)源是否連接成功，并動(dòng)態(tài)管理;
• 判斷數(shù)據(jù)源提供的賬號(hào)是否有操作權(quán)限，例如讀寫(xiě);
• 讀取數(shù)據(jù)源A的表結(jié)構(gòu)，在數(shù)據(jù)源B創(chuàng)建表;
• 數(shù)據(jù)讀取或者分頁(yè)讀取，寫(xiě)入數(shù)據(jù)源B中;
• 在不知道表結(jié)構(gòu)情況下，還需要讀取表結(jié)構(gòu)，生成SQL;

3、JDBC基礎(chǔ)API

• Statement

Java中JDBC下執(zhí)行數(shù)據(jù)庫(kù)操作的一個(gè)重要接口，在已經(jīng)建立數(shù)據(jù)庫(kù)連接的基礎(chǔ)上，向數(shù)據(jù)庫(kù)發(fā)送要執(zhí)行的SQL語(yǔ)句。

• PreparedStatement

繼承Statement接口，且實(shí)現(xiàn)SQL預(yù)編譯，可以提高批量處理效率。常應(yīng)用于批量數(shù)據(jù)寫(xiě)入場(chǎng)景。

• ResultSet

存儲(chǔ)JDBC查詢結(jié)果集的對(duì)象，ResultSet接口提供從當(dāng)前行檢索列值的方法。

二、基礎(chǔ)工具封裝

1、數(shù)據(jù)源管理

提供一個(gè)數(shù)據(jù)源管理的Factory，當(dāng)前場(chǎng)景下主要管理一個(gè)讀庫(kù)即數(shù)據(jù)源A，和一個(gè)寫(xiě)庫(kù)即數(shù)據(jù)源B，數(shù)據(jù)源連接驗(yàn)證通過(guò)，放入容器中。

@Component public class ConnectionFactory {private volatile Map<String, Connection> connectionMap = new HashMap<>();@Resourceprivate JdbcConfig jdbcConfig ;@PostConstructpublic 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、動(dòng)態(tài)SQL拼接

基礎(chǔ)SQL管理

主要提供SQL的基礎(chǔ)模板，例如全表查，分頁(yè)查，表結(jié)構(gòu)查詢。

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參數(shù)拼接

根據(jù)SQL模板中缺失的參數(shù)，進(jìn)行動(dòng)態(tài)補(bǔ)全，生成完成SQL語(yǔ)句。

public class BuildSql {/*** 讀權(quán)限SQL*/public static String buildReadSql(String table) {String readSql = null ;if (StringUtils.isNotEmpty(table)){readSql = String.format(BaseSql.READ_SQL, table);}return readSql;}/*** 讀權(quán)限SQL*/public static String buildWriteSql(String table){String writeSql = null ;if (StringUtils.isNotEmpty(table)){writeSql = String.format(BaseSql.WRITE_SQL, table,table);}return writeSql ;}/*** 表創(chuàng)建SQL*/public static String buildStructSql (String table){String structSql = null ;if (StringUtils.isNotEmpty(table)){structSql = String.format(BaseSql.CREATE_SQL, table);}return structSql ;}/*** 表結(jié)構(gòu)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 ;}/*** 總數(shù)查詢SQL*/public static String buildCountSql (String table){String countSql = null ;if (StringUtils.isNotEmpty(table)){countSql = String.format(BaseSql.COUNT_SQL,table);}return countSql ;}/*** 分頁(yè)查詢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 ;} }

三、業(yè)務(wù)化流程

1、基礎(chǔ)鑒權(quán)

讀庫(kù)嘗試一次單條數(shù)據(jù)讀取，寫(xiě)庫(kù)嘗試一次不成立條件的寫(xiě)入，如果沒(méi)有權(quán)限，會(huì)拋出相應(yīng)異常。

@RestController public class CheckController {@Resourceprivate 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、同步表結(jié)構(gòu)

這里執(zhí)行最簡(jiǎn)單操作，把讀庫(kù)表創(chuàng)建語(yǔ)句查詢出來(lái)，丟到寫(xiě)庫(kù)中執(zhí)行。

@RestController public class StructController {@Resourceprivate 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、同步表數(shù)據(jù)

讀庫(kù)的表數(shù)據(jù)讀取，批量放入寫(xiě)庫(kù)中。這里特別說(shuō)一個(gè)方法：statement.setObject();在不知道參數(shù)個(gè)數(shù)和類(lèi)型時(shí)，自動(dòng)適配數(shù)據(jù)類(lèi)型。

@RestController public class DataSyncController {@Resourceprivate 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(?,?)");// 基于動(dòng)態(tài)獲取列，和statement.setObject();自動(dòng)適配數(shù)據(jù)類(lèi)型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() ;}}// 處理最后一批數(shù)據(jù)statement.executeBatch();writeConnection.commit();}return rwReadEntities ;} catch (SQLException e) {e.printStackTrace();}return null ;} }

4、分頁(yè)查詢

提供一個(gè)分頁(yè)查詢工具，在數(shù)據(jù)量大的情況下不能一次性讀取大量的數(shù)據(jù)，避免資源占用過(guò)高。

public class PageUtilEntity {/*** 分頁(yè)生成方法*/public static PageHelperEntity<Object> pageResult (int total, int pageSize,int currentPage, List dataList){PageHelperEntity<Object> pageBean = new PageHelperEntity<Object>();// 總頁(yè)數(shù)int totalPage = PageHelperEntity.countTotalPage(pageSize,total) ;// 分頁(yè)列表List<Integer> pageList = PageHelperEntity.pageList(currentPage,pageSize,total) ;// 上一頁(yè)int prevPage = 0 ;if (currentPage==1){prevPage = currentPage ;} else if (currentPage>1&&currentPage<=totalPage){prevPage = currentPage -1 ;}// 下一頁(yè)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 ;} }

四、最后總結(jié)

很多復(fù)雜度偏高的業(yè)務(wù)，越是需要借助基礎(chǔ)API解決，因?yàn)閺?fù)雜度高，不容易抽象化統(tǒng)一封裝，如果數(shù)據(jù)同步這塊業(yè)務(wù)，可以適配多種數(shù)據(jù)庫(kù)，完全可以獨(dú)立封裝為中間件，開(kāi)源項(xiàng)目中關(guān)于多方數(shù)據(jù)同步或計(jì)算的中間件也有好多，可以自行了解下，增長(zhǎng)眼界開(kāi)闊思路。

五、源代碼地址

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

推薦相關(guān)閱讀

數(shù)據(jù)源管理：主從庫(kù)動(dòng)態(tài)路由，AOP模式讀寫(xiě)分離

數(shù)據(jù)源管理：基于JDBC模式，適配和管理動(dòng)態(tài)數(shù)據(jù)源

總結(jié)

以上是生活随笔為你收集整理的数据源管理 | 动态权限校验，表结构和数据迁移流程的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。