shardingsphere通過注解實現分庫

1、前言

    上篇文章大概講了下shardingSphere中的sharding-jdbc的實現原理（https://www.cnblogs.com/smileIce/p/11131053.html），接下來我們想正對大家使用場景來分析下。

    大家可以先看看shardingSphere的配置方式https://shardingsphere.apache.org/document/current/cn/manual/sharding-jdbc/configuration/config-spring-boot/。

    我們可以發現shardingSphere是支持默認的 default-database-strategy或者針對表上的default-database-strategy。

    對於規模不太大的公司來說，針對每個表配置default-database-strategy工作量還是比較大的，一般情況會針對不同的業務使用多個庫，一個工程可能會使用多個庫的數據（可能有人會說，用微服務不應該有一個項目使用多個庫的情況，但是在真實的業務場景確實可能一個項目需要調用多個庫，可能為了性能考慮，可能有其它業務方面的考慮）。

    針對該場景可能通過注解的方式更好。用spring  AbstractRoutingDataSource 方式線下如下 https://juejin.im/post/5a927d23f265da4e7e10d740，但是該方式針對事務就無能為力（針對分布式事務后面會講到）。那么sharding-sphere是否能實現通過配置一個注解就能簡單的切換數據源昵，當然可以了，接下來看看怎么實現。

 

2、實現annotation的數據源原理。

    上一節提到HintManager，但沒有細講，這節可以講講HintManager

    

public final class HintManager implements AutoCloseable {
    
    private static final ThreadLocal<HintManager> HINT_MANAGER_HOLDER = new ThreadLocal<>();
    
    private final Multimap<String, Comparable<?>> databaseShardingValues = HashMultimap.create();
    
    private final Multimap<String, Comparable<?>> tableShardingValues = HashMultimap.create();
    
    private boolean databaseShardingOnly;
    
    private boolean masterRouteOnly;
    
    /**
     * Get a new instance for {@code HintManager}.
     *
     * @return  {@code HintManager} instance
     */
    public static HintManager getInstance() {
        Preconditions.checkState(null == HINT_MANAGER_HOLDER.get(), "Hint has previous value, please clear first.");
        HintManager result = new HintManager();
        HINT_MANAGER_HOLDER.set(result);
        return result;
    }
    
    /**
     * Set sharding value for database sharding only.
     *
     * <p>The sharding operator is {@code =}</p>
     *
     * @param value sharding value
     */
    public void setDatabaseShardingValue(final Comparable<?> value) {
        databaseShardingValues.clear();
        databaseShardingValues.put("", value);
        databaseShardingOnly = true;
    }
    
    /**
     * Add sharding value for database.
     *
     * <p>The sharding operator is {@code =}</p>
     *
     * @param logicTable logic table name
     * @param value sharding value
     */
    public void addDatabaseShardingValue(final String logicTable, final Comparable<?> value) {
        databaseShardingValues.put(logicTable, value);
        databaseShardingOnly = false;
    }
    
    /**
     * Add sharding value for table.
     *
     * <p>The sharding operator is {@code =}</p>
     *
     * @param logicTable logic table name
     * @param value sharding value
     */
    public void addTableShardingValue(final String logicTable, final Comparable<?> value) {
        tableShardingValues.put(logicTable, value);
        databaseShardingOnly = false;
    }
    
    /**
     * Get database sharding values.
     *
     * @return database sharding values
     */
    public static Collection<Comparable<?>> getDatabaseShardingValues() {
        return getDatabaseShardingValues("");
    }

    /**
     * Get database sharding values.
     *
     * @param logicTable logic table
     * @return database sharding values
     */
    public static Collection<Comparable<?>> getDatabaseShardingValues(final String logicTable) {
        return null == HINT_MANAGER_HOLDER.get() ? Collections.<Comparable<?>>emptyList() : HINT_MANAGER_HOLDER.get().databaseShardingValues.get(logicTable);
    }
    
    /**
     * Get table sharding values.
     *
     * @param logicTable logic table name
     * @return table sharding values
     */
    public static Collection<Comparable<?>> getTableShardingValues(final String logicTable) {
        return null == HINT_MANAGER_HOLDER.get() ? Collections.<Comparable<?>>emptyList() : HINT_MANAGER_HOLDER.get().tableShardingValues.get(logicTable);
    }
    
    /**
     * Judge whether database sharding only.
     *
     * @return database sharding or not
     */
    public static boolean isDatabaseShardingOnly() {
        return null != HINT_MANAGER_HOLDER.get() && HINT_MANAGER_HOLDER.get().databaseShardingOnly;
    }
    
    /**
     * Set database operation force route to master database only.
     */
    public void setMasterRouteOnly() {
        masterRouteOnly = true;
    }
    
    /**
     * Judge whether route to master database only or not.
     *
     * @return route to master database only or not
     */
    public static boolean isMasterRouteOnly() {
        return null != HINT_MANAGER_HOLDER.get() && HINT_MANAGER_HOLDER.get().masterRouteOnly;
    }
    
    /**
     * Clear threadlocal for hint manager.
     */
    public static void clear() {
        HINT_MANAGER_HOLDER.remove();
    }
    
    @Override
    public void close() {
        HintManager.clear();
    }
}

   通過源碼我可以發現HintManager 維護了一個線程變量 HINT_MANAGER_HOLDER，我們這里就是通過該變量實現的注解分庫。

還記得databaseShardingOnly這個變量嗎？這個變量決定了是使用DatabaseHintSQLRouter還是使用ParsingSQLRouter

public final class ShardingRouterFactory {
    
    /**
     * Create new instance of sharding router.
     * 
     * @param shardingRule sharding rule
     * @param shardingMetaData sharding meta data
     * @param databaseType database type
     * @param showSQL show SQL or not
     * @return sharding router instance
     */
    public static ShardingRouter newInstance(final ShardingRule shardingRule, final ShardingMetaData shardingMetaData, final DatabaseType databaseType, final boolean showSQL) {
        return HintManagerHolder.isDatabaseShardingOnly() ? new DatabaseHintSQLRouter(shardingRule, showSQL) : new ParsingSQLRouter(shardingRule, shardingMetaData, databaseType, showSQL);
    }
}

         我們只要通過調用HintManager 的setDatabaseShardingValue，我們就可以設置為只進行分庫並把分庫的數據Id 填入其中

      

   public void setDatabaseShardingValue(final Comparable<?> value) {

        databaseShardingValues.clear();
        databaseShardingValues.put("", value);
        databaseShardingOnly = true;
    }

 

  3、實現邏輯

定義數據庫類型

public enum DataSourceType {

    DEFAULT("default", "default"),

    DATASOURCE_1("datasource1", "datasource1"),

    DATASOURCE_2("datasource1", "datasource2"),

    ;

    private String name;

    private String identity;

    DataSourceType(String name, String identity) {
        this.name = name;
        this.identity = identity;
    }

    public String getName() {
        return name;
    }

    public String getIdentity() {
        return identity;
    }

    public static DataSourceType getDataSourceTypeByName(String name) {
        for (DataSourceType dataSourceType : DataSourceType.values()) {
            if (dataSourceType.name.equals(name)) {
                return dataSourceType;
            }
        }
        throw new RuntimeException("db is not exist." + name);
    }
}

 

 

定義注解類

@Target({ElementType.METHOD, ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface TargetDataSource {

    DataSourceType value();

    boolean isDatabaseShardingOnly() default true;
}

 

定義AOP攔截器

@Aspect
@Order(-10)
@Component

public class DynamicDataSourceAspect {


    private static final Logger LOGGER = LogManager.getLogger(DynamicDataSourceAspect.class);


    @Before("within(@com.newretail.datasource.TargetDataSource *) || @annotation(com.newretail.datasource.TargetDataSource)")
    public void changeDataSource(JoinPoint point) {
        MethodSignature joinPointObject = (MethodSignature) point.getSignature();
        TargetDataSource targetDataSource = null;
        if (joinPointObject.getDeclaringType().isAnnotationPresent(TargetDataSource.class)) {
            targetDataSource = (TargetDataSource) joinPointObject.getDeclaringType().getAnnotation(TargetDataSource.class);
        }
        Method method = joinPointObject.getMethod();
        if (method.isAnnotationPresent(TargetDataSource.class)) {
            targetDataSource = method.getAnnotation(TargetDataSource.class);
        }
        if (targetDataSource.isDatabaseShardingOnly()) {
            //獲取當前的指定的數據源;
            DataSourceType dsId = targetDataSource.value();
            HintManager.getInstance().setDatabaseShardingValue(dsId.getIdentity());
        }

    }


    @After(value = "@annotation(com.newretail.datasource.TargetDataSource)")
    public void restoreDataSource(JoinPoint point) {

        //方法執行完畢之后，銷毀當前數據源信息，進行垃圾回收。
        HintManagerHolder.clear();

    }


}

 

實現HintShardingAlgorithm算法。

public class AnnotationHintShardingAlgorithm implements HintShardingAlgorithm {


    private final static Logger logger = LoggerFactory.getLogger(AnnotationHintShardingAlgorithm.class);

    @Override
    public Collection<String> doSharding(Collection<String> availableTargetNames, ShardingValue shardingValue) {
        if (shardingValue instanceof ListShardingValue && !CollectionUtils.isEmpty(((ListShardingValue) shardingValue).getValues())) {
            logger.info("---------------------"+((ListShardingValue) shardingValue).getValues());
            return availableTargetNames.stream().filter(availableTargetName ->
                    ((ListShardingValue) shardingValue).getValues().contains(availableTargetName)).collect(Collectors.toList());
        }
        return availableTargetNames;

    }
}

 

配置如下

# 數據源 load.datasources 可以指定需要加載的數據源，沒配置加載所有的
sharding:
  jdbc:
    datasource:
      names:
        datasource1,datasource2
      datasource1:
        type: com.zaxxer.hikari.HikariDataSource 
        driver-class-name: com.mysql.cj.jdbc.Driver 
        jdbcUrl: 

 username: 

 password:  datasource2: type: com.zaxxer.hikari.HikariDataSource driver-class-name: com.mysql.cj.jdbc.Driver jdbcUrl:   username: password:  config: sharding: defaultDatabaseStrategy: hint: algorithmClassName: com.newretail.datasource.AnnotationHintShardingAlgorithm props: sql: show: false

 

 

public interface DeliveryTargetMapper {

    @TargetDataSource(value = DataSourceType.DATASOURCE_1)
    List<City> getCityList();

}