資料來源管理 | 動態許可權校驗,表結構和資料遷移流程
本文原始碼:GitHub·點這裡 || GitEE·點這裡
一、資料同步簡介
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&¤tPage<=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
推薦相關閱讀 |
---|
資料來源管理:主從庫動態路由,AOP模式讀寫分離 |
資料來源管理:基於JDBC模式,適配和管理動態資料來源 |