SpringBoot實現分庫分表
方案:可以使用攔截器攔截mybatis框架,在執行SQL前對SQL語句根據路由字段進行分庫分表操作,下例隻做分表功能
@Intercepts:申明需要攔截的方法
攔截StatementHandler對象
一、statementHandler對象的定義
首先我們先來看看statementHandler接口的定義:
首先約定文中將的四大對象是指:executor, statementHandler,parameterHandler,resultHandler對象。
SimpleStatementHandler
:對應我們JDBC中常用的Statement接口,用於簡單SQL的處理;PreparedStatementHandler
:對應JDBC中的PreparedStatement,預編譯SQL的接口;CallableStatementHandler
:對應JDBC中CallableStatement,用於執行存儲過程相關的接口;RoutingStatementHandler
:這個接口是以上三個接口的路由,沒有實際操作,隻是負責上面三個StatementHandler的創建及調用。
講到statementHandler,毫無疑問它是我們四大對象最重要的一個,它的任務就是和數據庫對話。在它這裡會使用parameterHandler和ResultHandler對象為我們綁定SQL參數和組裝最後的結果返回。
public interface StatementHandler { Statement prepare(Connection connection) throws SQLException; void parameterize(Statement statement) throws SQLException; void batch(Statement statement) throws SQLException; int update(Statement statement) throws SQLException; <E> List<E> query(Statement statement, ResultHandler resultHandler) throws SQLException; BoundSql getBoundSql(); ParameterHandler getParameterHandler(); }
二、prepare方法
1、首先prepare方法是用來編譯SQL
讓我們看看它的源碼實現。這裡我們看到瞭BaseStatementHandler對prepare方法的實現
@Override public Statement prepare(Connection connection) throws SQLException { ErrorContext.instance().sql(boundSql.getSql()); Statement statement = null; try { statement = instantiateStatement(connection); setStatementTimeout(statement); setFetchSize(statement); return statement; } catch (SQLException e) { closeStatement(statement); throw e; } catch (Exception e) { closeStatement(statement); throw new ExecutorException("Error preparing statement. Cause: " + e, e); } } protected abstract Statement instantiateStatement(Connection connection) throws SQLException;
顯然我們通過源碼更加關註抽象方法instantiateStatement是做瞭什麼事情。它依舊是一個抽象方法,那麼它就有其實現類。
2、那就是之前說的那幾個具體的StatementHandler對象
讓我們看看PreparedStatementHandler:
@Override protected Statement instantiateStatement(Connection connection) throws SQLException { String sql = boundSql.getSql(); if (mappedStatement.getKeyGenerator() instanceof Jdbc3KeyGenerator) { String[] keyColumnNames = mappedStatement.getKeyColumns(); if (keyColumnNames == null) { return connection.prepareStatement(sql, PreparedStatement.RETURN_GENERATED_KEYS); } else { return connection.prepareStatement(sql, keyColumnNames); } } else if (mappedStatement.getResultSetType() != null) { return connection.prepareStatement(sql, mappedStatement.getResultSetType().getValue(), ResultSet.CONCUR_READ_ONLY); } else { return connection.prepareStatement(sql); } }
好這個方法非常簡單,我們可以看到它主要是根據上下文來預編譯SQL,這是我們還沒有設置參數。設置參數的任務是交由,statement接口的parameterize方法來實現的。
3、parameterize方法
上面我們在prepare方法裡面預編譯瞭SQL。那麼我們這個時候希望設置參數。在Statement中我們是使用parameterize方法進行設置參數的。
讓我們看看PreparedStatementHandler中的parameterize方法:
@Override public void parameterize(Statement statement) throws SQLException { parameterHandler.setParameters((PreparedStatement) statement); }
很顯然這裡很簡單是通過parameterHandler來實現的,我們這篇文章隻是停留在statementhandler的程度,等我們講解parameterHandler的時候再來看它如何實現吧,期待一下吧。
4、query/update方法
我們用瞭prepare方法預編譯瞭SQL,用瞭parameterize方法設置參數,那麼我們接下來肯定是想執行SQL,而SQL無非是兩種:
一種是進行查詢——query,另外就是更新——update。
這些方法都很簡單,讓我們看看PreparedStatementHandler的實現:
@Override public int update(Statement statement) throws SQLException { PreparedStatement ps = (PreparedStatement) statement; ps.execute(); int rows = ps.getUpdateCount(); Object parameterObject = boundSql.getParameterObject(); KeyGenerator keyGenerator = mappedStatement.getKeyGenerator(); keyGenerator.processAfter(executor, mappedStatement, ps, parameterObject); return rows; } @Override public <E> List<E> query(Statement statement, ResultHandler resultHandler) throws SQLException { PreparedStatement ps = (PreparedStatement) statement; ps.execute(); return resultSetHandler.<E> handleResultSets(ps); }
例:動態替換SQL中@TableID標識符
package com.study.demo.interceptor; import com.study.demo.exception.BaseException; import org.apache.ibatis.executor.statement.StatementHandler; import org.apache.ibatis.mapping.BoundSql; import org.apache.ibatis.plugin.Interceptor; import org.apache.ibatis.plugin.Intercepts; import org.apache.ibatis.plugin.Invocation; import org.apache.ibatis.plugin.Plugin; import org.apache.ibatis.plugin.Signature; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Component; import org.springframework.util.CollectionUtils; import java.lang.reflect.Field; import java.sql.Connection; import java.util.Map; import java.util.Properties; import java.util.Set; @Component @Intercepts({ @Signature(type = StatementHandler.class, method = "prepare", args = {Connection.class, Integer.class})}) public class DynamicSQLInterceptor implements Interceptor { private static final Logger LOGGER = LoggerFactory.getLogger(DynamicSQLInterceptor.class); private static final String SHARD_TABLE_ID = "SHARD_TABLE_ID"; private static final String DEFAULT_TABLE_ID = "000"; @Override @SuppressWarnings("unchecked") public Object intercept(Invocation invocation) throws Throwable { LOGGER.info("DynamicSQLInterceptor.intercept() exec."); StatementHandler statementHandler = (StatementHandler) invocation.getTarget(); Object parameter = statementHandler.getParameterHandler().getParameterObject(); Map<String, Object> params = (Map)parameter; if(CollectionUtils.isEmpty(params)){ throw new BaseException("SQL: 路由字段不能為空!"); } String tableId = DEFAULT_TABLE_ID; Set<String> keySet = params.keySet(); for (String key : keySet) { if (SHARD_TABLE_ID.equals(key)) { tableId = String.valueOf(params.get(key)); } } BoundSql boundSql = statementHandler.getBoundSql(); //獲取到原始sql語句 String sql = boundSql.getSql(); String newSql = sql.replaceAll("@TableID", tableId); LOGGER.debug("[DynamicSQLInterceptor] Sql:{}", newSql); //通過反射修改sql語句 Field field = boundSql.getClass().getDeclaredField("sql"); field.setAccessible(true); field.set(boundSql, newSql); return invocation.proceed(); } @Override public Object plugin(Object target) { //隻攔截Executor對象,減少目標被代理的次數 if (target instanceof StatementHandler) { return Plugin.wrap(target, this); } else { return target; } } @Override public void setProperties(Properties properties) { LOGGER.debug("[DynamicSQLInterceptor] SetProperties"); } }
示例SQL:
SELECT * FROM ST_CLASS_@TableID WHERE ID = #{id}
service層示例:
@Override public Objcet queryByPrimaryKey(String id) { Map<String, Object> params = DbShardUtils.shardDBParamMap(id); params.put("id", id); return testDao.queryByPrimaryKey(params); }
dao層示例:
@Repository public interface TestDao { Object queryByPrimaryKey(Map<String, Object> params); }
package com.study.demo.utils; import org.apache.commons.lang3.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.HashMap; import java.util.Map; /** * 分庫分表工具類 <br> * 返回Map<String, Object>, 含有key:SHARD_TABLE_ID */ public class DbShardUtils { private static final Logger LOGGER = LoggerFactory.getLogger(DbShardUtils.class); private static final String SHARD_TABLE_ID = "SHARD_TABLE_ID"; /** * 私有構造函數 */ private DbShardUtils() { } public static Map<String, Object> shardDBParamMap(String id){ if (StringUtils.isBlank(id)) { LOGGER.error("sharding id is null"); } Map<String, Object> paramMap = new HashMap<>(); paramMap.put(SHARD_TABLE_ID, rout(id)); return paramMap; } private static String rout(String id) { // 測試 return "000"; } }
以上為個人經驗,希望能給大傢一個參考,也希望大傢多多支持WalkonNet。
推薦閱讀:
- mybatis源碼解讀之executor包語句處理功能
- MyBatis攔截器的實現原理
- Mybatis 插件原理解析
- Mybatis執行Update返回行數為負數的問題
- 淺談Mybatis SqlSession執行流程