文章目录
- 主键生成策略源码
- KeyGenerateAlgorithm
- 源码入口
- 实现
- 扩展 自定义分布式主键生成策略
- 分片算法
- ShardingAlgorithm
- 实现
- 扩展 自定义分片算法
- 踩的坑
主键生成策略源码
开发者手册
KeyGenerateAlgorithm
全限定类名org.apache.shardingsphere.sharding.spi.KeyGenerateAlgorithm
分布式主键生成算法,已知实现
配置标识 | 详细说明 | 全限定类名 |
---|---|---|
SNOWFLAKE | 基于雪花算法的分布式主键生成算法 | org.apache.shardingsphere.sharding.algorithm.keygen.SnowflakeKeyGenerateAlgorithm |
UUID | 基于 UUID 的分布式主键生成算法 | org.apache.shardingsphere.sharding.algorithm.keygen.UUIDKeyGenerateAlgorithm |
NANOID | 基于 NanoId 的分布式主键生成算法 | org.apache.shardingsphere.sharding.nanoid.algorithm.keygen.NanoIdKeyGenerateAlgorithm |
COSID | 基于 CosId 的分布式主键生成算法 | org.apache.shardingsphere.sharding.cosid.algorithm.keygen.CosIdKeyGenerateAlgorithm |
COSID_SNOWFLAKE | 基于 CosId 的雪花算法分布式主键生成算法 | org.apache.shardingsphere.sharding.cosid.algorithm.keygen.CosIdSnowflakeKeyGenerateAlgorithm |
源码入口
package org.apache.shardingsphere.sharding.factory;/*** Key generate algorithm factory.*/
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public final class KeyGenerateAlgorithmFactory {//加载所有的主键生成策略static {ShardingSphereServiceLoader.register(KeyGenerateAlgorithm.class);}/*** 根据配置的主键生成策略,获取一个主键生成算法* 例如:spring.shardingsphere.rules.sharding.key-generators.usercourse_keygen.type=SNOWFLAKE*/public static KeyGenerateAlgorithm newInstance(final AlgorithmConfiguration keyGenerateAlgorithmConfig) {return ShardingSphereAlgorithmFactory.createAlgorithm(keyGenerateAlgorithmConfig, KeyGenerateAlgorithm.class);}/*** 判断是否包含配置的算法*/public static boolean contains(final String keyGenerateAlgorithmType) {return TypedSPIRegistry.findRegisteredService(KeyGenerateAlgorithm.class, keyGenerateAlgorithmType).isPresent();}
}
先来看主键生成策略是如何加载的:ShardingSphereServiceLoader.register(KeyGenerateAlgorithm.class);
public final class ShardingSphereServiceLoader {//线程安全Map,缓存所有主键生成器private static final Map<Class<?>, Collection<Object>> SERVICES = new ConcurrentHashMap<>();// 进入到register()方法中public static void register(final Class<?> serviceInterface) {if (!SERVICES.containsKey(serviceInterface)) {// 调用下方的load()方法SERVICES.put(serviceInterface, load(serviceInterface));}}//使用java的SPI机制加载接口的所有实现类private static <T> Collection<Object> load(final Class<T> serviceInterface) {Collection<Object> result = new LinkedList<>();for (T each : ServiceLoader.load(serviceInterface)) {result.add(each);}return result;}
}
实现
ShardingJDBC是通过SPI机制,加载org.apache.shardingsphere.sharding.spi.KeyGenerateAlgorithm
接口的实现类,也就是上方表格中的内容
我们就可以直接在yml配置文件中进行配置分布式主键生成算法
接下来就以SNOWFLAKE雪花算法举例,下方就列举出了几个关键方法
// 实现了KeyGenerateAlgorithm接口
public final class SnowflakeKeyGenerateAlgorithm implements KeyGenerateAlgorithm, InstanceContextAware {// 在init方法中,会把我们yml配置文件中定义的props的配置项,保存在下面方法的形参中,并赋值给props成员属性// 其他地方再用props对象获取我们的配置项@Overridepublic void init(final Properties props) {this.props = props;maxVibrationOffset = getMaxVibrationOffset(props);maxTolerateTimeDifferenceMilliseconds = getMaxTolerateTimeDifferenceMilliseconds(props);}// 实现KeyGenerateAlgorithm接口中的抽象方法generateKey()// 也就是在这个方法中具体生成分布式主键值的@Overridepublic synchronized Long generateKey() {long currentMilliseconds = timeService.getCurrentMillis();if (waitTolerateTimeDifferenceIfNeed(currentMilliseconds)) {currentMilliseconds = timeService.getCurrentMillis();}if (lastMilliseconds == currentMilliseconds) {if (0L == (sequence = (sequence + 1) & SEQUENCE_MASK)) {currentMilliseconds = waitUntilNextTime(currentMilliseconds);}} else {vibrateSequenceOffset();sequence = sequenceOffset;}lastMilliseconds = currentMilliseconds;return ((currentMilliseconds - EPOCH) << TIMESTAMP_LEFT_SHIFT_BITS) | (getWorkerId() << WORKER_ID_LEFT_SHIFT_BITS) | sequence;}// getType() 方法中返回的字符串就是我们上方yml配置文件中type配置项填写的值@Overridepublic String getType() {return "SNOWFLAKE";}
}
其他几个实现类也是一样的格式
扩展 自定义分布式主键生成策略
package com.hs.sharding.algorithm;import com.google.common.base.Preconditions;
import org.apache.shardingsphere.infra.instance.InstanceContext;
import org.apache.shardingsphere.infra.instance.InstanceContextAware;
import org.apache.shardingsphere.sharding.algorithm.keygen.TimeService;
import org.apache.shardingsphere.sharding.spi.KeyGenerateAlgorithm;import java.util.Calendar;
import java.util.Properties;/*** 改进雪花算法,让他能够 %4 均匀分布。* @auth hs*/
public final class MySnowFlakeAlgorithm implements KeyGenerateAlgorithm, InstanceContextAware {public static final long EPOCH;private static final String MAX_VIBRATION_OFFSET_KEY = "max-vibration-offset";private static final String MAX_TOLERATE_TIME_DIFFERENCE_MILLISECONDS_KEY = "max-tolerate-time-difference-milliseconds";private static final long SEQUENCE_BITS = 12L;private static final long WORKER_ID_BITS = 10L;private static final long SEQUENCE_MASK = (1 << SEQUENCE_BITS) - 1;private static final long WORKER_ID_LEFT_SHIFT_BITS = SEQUENCE_BITS;private static final long TIMESTAMP_LEFT_SHIFT_BITS = WORKER_ID_LEFT_SHIFT_BITS + WORKER_ID_BITS;private static final int DEFAULT_VIBRATION_VALUE = 1;private static final int MAX_TOLERATE_TIME_DIFFERENCE_MILLISECONDS = 10;private static final long DEFAULT_WORKER_ID = 0;private static TimeService timeService = new TimeService();public static void setTimeService(TimeService timeService) {MySnowFlakeAlgorithm.timeService = timeService;}private Properties props;@Overridepublic Properties getProps() {return props;}private int maxVibrationOffset;private int maxTolerateTimeDifferenceMilliseconds;private volatile int sequenceOffset = -1;private volatile long sequence;private volatile long lastMilliseconds;private volatile InstanceContext instanceContext;static {Calendar calendar = Calendar.getInstance();calendar.set(2016, Calendar.NOVEMBER, 1);calendar.set(Calendar.HOUR_OF_DAY, 0);calendar.set(Calendar.MINUTE, 0);calendar.set(Calendar.SECOND, 0);calendar.set(Calendar.MILLISECOND, 0);EPOCH = calendar.getTimeInMillis();}@Overridepublic void init(final Properties props) {this.props = props;maxVibrationOffset = getMaxVibrationOffset(props);maxTolerateTimeDifferenceMilliseconds = getMaxTolerateTimeDifferenceMilliseconds(props);}@Overridepublic void setInstanceContext(final InstanceContext instanceContext) {this.instanceContext = instanceContext;if (null != instanceContext) {instanceContext.generateWorkerId(props);}}private int getMaxVibrationOffset(final Properties props) {int result = Integer.parseInt(props.getOrDefault(MAX_VIBRATION_OFFSET_KEY, DEFAULT_VIBRATION_VALUE).toString());Preconditions.checkArgument(result >= 0 && result <= SEQUENCE_MASK, "Illegal max vibration offset.");return result;}private int getMaxTolerateTimeDifferenceMilliseconds(final Properties props) {return Integer.parseInt(props.getOrDefault(MAX_TOLERATE_TIME_DIFFERENCE_MILLISECONDS_KEY, MAX_TOLERATE_TIME_DIFFERENCE_MILLISECONDS).toString());}@Overridepublic synchronized Long generateKey() {long currentMilliseconds = timeService.getCurrentMillis();if (waitTolerateTimeDifferenceIfNeed(currentMilliseconds)) {currentMilliseconds = timeService.getCurrentMillis();}if (lastMilliseconds == currentMilliseconds) {
// if (0L == (sequence = (sequence + 1) & SEQUENCE_MASK)) {currentMilliseconds = waitUntilNextTime(currentMilliseconds);
// }} else {vibrateSequenceOffset();
// sequence = sequenceOffset;sequence = sequence >= SEQUENCE_MASK ? 0:sequence+1;}lastMilliseconds = currentMilliseconds;return ((currentMilliseconds - EPOCH) << TIMESTAMP_LEFT_SHIFT_BITS) | (getWorkerId() << WORKER_ID_LEFT_SHIFT_BITS) | sequence;}private boolean waitTolerateTimeDifferenceIfNeed(final long currentMilliseconds) {if (lastMilliseconds <= currentMilliseconds) {return false;}long timeDifferenceMilliseconds = lastMilliseconds - currentMilliseconds;Preconditions.checkState(timeDifferenceMilliseconds < maxTolerateTimeDifferenceMilliseconds,"Clock is moving backwards, last time is %d milliseconds, current time is %d milliseconds", lastMilliseconds, currentMilliseconds);try {Thread.sleep(timeDifferenceMilliseconds);} catch (InterruptedException e) {}return true;}private long waitUntilNextTime(final long lastTime) {long result = timeService.getCurrentMillis();while (result <= lastTime) {result = timeService.getCurrentMillis();}return result;}@SuppressWarnings("NonAtomicOperationOnVolatileField")private void vibrateSequenceOffset() {sequenceOffset = sequenceOffset >= maxVibrationOffset ? 0 : sequenceOffset + 1;}private long getWorkerId() {return null == instanceContext ? DEFAULT_WORKER_ID : instanceContext.getWorkerId();}@Overridepublic String getType() {return "MYSNOWFLAKE";}@Overridepublic boolean isDefault() {return true;}
}
使用spi机制加载我们上方定义的类
yml配置文件中使用我们自己定义的类
分片算法
开发者手册
ShardingAlgorithm
全限定类名org.apache.shardingsphere.sharding.spi.ShardingAlgorithm
分片算法,已知实现
配置标识 | 自动分片算法 | 详细说明 | 类名 |
---|---|---|---|
MOD | Y | 基于取模的分片算法 | ModShardingAlgorithm |
HASH_MOD | Y | 基于哈希取模的分片算法 | HashModShardingAlgorithm |
BOUNDARY_RANGE | Y | 基于分片边界的范围分片算法 | BoundaryBasedRangeShardingAlgorithm |
VOLUME_RANGE | Y | 基于分片容量的范围分片算法 | VolumeBasedRangeShardingAlgorithm |
AUTO_INTERVAL | Y | 基于可变时间范围的分片算法 | AutoIntervalShardingAlgorithm |
INTERVAL | N | 基于固定时间范围的分片算法 | IntervalShardingAlgorithm |
CLASS_BASED | N | 基于自定义类的分片算法 | ClassBasedShardingAlgorithm |
INLINE | N | 基于行表达式的分片算法 | InlineShardingAlgorithm |
COMPLEX_INLINE | N | 基于行表达式的复合分片算法 | ComplexInlineShardingAlgorithm |
HINT_INLINE | N | 基于行表达式的 Hint 分片算法 | HintInlineShardingAlgorithm |
COSID_MOD | N | 基于 CosId 的取模分片算法 | CosIdModShardingAlgorithm |
COSID_INTERVAL | N | 基于 CosId 的固定时间范围的分片算法 | CosIdIntervalShardingAlgorithm |
COSID_INTERVAL_SNOWFLAKE | N | 基于 CosId 的雪花ID固定时间范围的分片算法 | CosIdSnowflakeIntervalShardingAlgorithm |
实现
这里就拿CLASS_BASED自定义分片策略来举例。我们之前的配置项如下所示。
这里就有一个问题,props的值我怎么知道写什么,我又怎么知道我自定义的类需要实现什么接口?
我们现在进入到CLASS_BASED分片算法的实现类中ClassBasedShardingAlgorithm
去看看它的源码
public final class ClassBasedShardingAlgorithm implements StandardShardingAlgorithm<Comparable<?>>, ComplexKeysShardingAlgorithm<Comparable<?>>, HintShardingAlgorithm<Comparable<?>> {// 定义两个常量,我们会发现这里就是props中我们进行配置的值private static final String STRATEGY_KEY = "strategy";private static final String ALGORITHM_CLASS_NAME_KEY = "algorithmClassName";@Getterprivate Properties props;private ClassBasedShardingAlgorithmStrategyType strategy;private String algorithmClassName;private StandardShardingAlgorithm standardShardingAlgorithm;private ComplexKeysShardingAlgorithm complexKeysShardingAlgorithm;private HintShardingAlgorithm hintShardingAlgorithm;// init()方法中会获取到props对象,props对象中保存了我们yml配置文件中的配置内容// 这里就会取出来,赋值给 strategy 和 algorithmClassName 成员属性@Overridepublic void init(final Properties props) {this.props = props;strategy = getStrategy(props);algorithmClassName = getAlgorithmClassName(props);initAlgorithmInstance(props);}private ClassBasedShardingAlgorithmStrategyType getStrategy(final Properties props) {String strategy = props.getProperty(STRATEGY_KEY);Preconditions.checkNotNull(strategy, "Properties `%s` can not be null when uses class based sharding strategy.", STRATEGY_KEY);return ClassBasedShardingAlgorithmStrategyType.valueOf(strategy.toUpperCase().trim());}private String getAlgorithmClassName(final Properties props) {String result = props.getProperty(ALGORITHM_CLASS_NAME_KEY);Preconditions.checkNotNull(result, "Properties `%s` can not be null when uses class based sharding strategy.", ALGORITHM_CLASS_NAME_KEY);return result;}// 这里就会判断 strategy 属性是哪一个 STANDARD、COMPLEX、HINT// 然后在进行具体的实例 StandardShardingAlgorithm、ComplexKeysShardingAlgorithm、HintShardingAlgorithmprivate void initAlgorithmInstance(final Properties props) {switch (strategy) {case STANDARD:standardShardingAlgorithm = ClassBasedShardingAlgorithmFactory.newInstance(algorithmClassName, StandardShardingAlgorithm.class, props);break;case COMPLEX:complexKeysShardingAlgorithm = ClassBasedShardingAlgorithmFactory.newInstance(algorithmClassName, ComplexKeysShardingAlgorithm.class, props);break;case HINT:hintShardingAlgorithm = ClassBasedShardingAlgorithmFactory.newInstance(algorithmClassName, HintShardingAlgorithm.class, props);break;default:break;}}// doSharding()方法,具体的分片算法逻辑@SuppressWarnings("unchecked")@Overridepublic String doSharding(final Collection<String> availableTargetNames, final PreciseShardingValue<Comparable<?>> shardingValue) {return standardShardingAlgorithm.doSharding(availableTargetNames, shardingValue);}@SuppressWarnings("unchecked")@Overridepublic Collection<String> doSharding(final Collection<String> availableTargetNames, final RangeShardingValue<Comparable<?>> shardingValue) {return standardShardingAlgorithm.doSharding(availableTargetNames, shardingValue);}@SuppressWarnings("unchecked")@Overridepublic Collection<String> doSharding(final Collection<String> availableTargetNames, final ComplexKeysShardingValue<Comparable<?>> shardingValue) {return complexKeysShardingAlgorithm.doSharding(availableTargetNames, shardingValue);}@SuppressWarnings("unchecked")@Overridepublic Collection<String> doSharding(final Collection<String> availableTargetNames, final HintShardingValue<Comparable<?>> shardingValue) {return hintShardingAlgorithm.doSharding(availableTargetNames, shardingValue);}// 返回trye为 CLASS_BASED 这里也就是和yml配置文件中的type对应上了@Overridepublic String getType() {return "CLASS_BASED";}
}
其他的分片算法也是类似的实现
扩展 自定义分片算法
自定义一个java类,实现ShardingAlgorithm接口,或者是它的子接口 StandardShardingAlgorithm、ComplexKeysShardingAlgorithm、HintShardingAlgorithm都行,重写其中的doSharding()
方法,我们自己指定分片逻辑
重写getType()
方法,返回一个字符串,能够让我们在yml配置文件中进行配置
@Override
public String getType() {return "MY_COMPLEX_ALGORITHM";
}
例如我现在自定义的分片类如下
package com.hs.sharding.algorithm;import com.google.common.base.Preconditions;
import com.google.common.collect.Range;
import org.apache.shardingsphere.sharding.api.sharding.standard.PreciseShardingValue;
import org.apache.shardingsphere.sharding.api.sharding.standard.RangeShardingValue;
import org.apache.shardingsphere.sharding.api.sharding.standard.StandardShardingAlgorithm;import java.util.*;/*** 自定义分片策略 , 我们这里实现标准的分片算法接口StandardShardingAlgorithm* 我这里是分片逻辑就是按照个数取模,在分发到sys_user1 sys_user2数据表中*/
public class HsComplexAlgorithm implements StandardShardingAlgorithm<Long> {/*** 数据库个数*/private final String DB_COUNT = "db-count";/*** 数据表个数*/private final String TAB_COUNT = "tab-count";/*** 真实数据表前缀*/private final String PERTAB = "pertab";private Integer dbCount;private Integer tabCount;private String pertab;private Properties props;@Overridepublic void init(Properties props) {this.props = props;this.dbCount = getDbCount(props);this.tabCount = getTabCount(props);this.pertab = getPertab(props);// 校验条件Preconditions.checkState(null != pertab && !pertab.isEmpty(),"Inline hsComplex algorithm expression cannot be null or empty.");}/*** 精确查询分片执行接口(对应的sql是where ??=值)* @param collection 可用的分片名集合(分库就是库名,分表就是表名)* @param preciseShardingValue 分片键*/@Overridepublic String doSharding(Collection<String> collection, PreciseShardingValue<Long> preciseShardingValue) {Long uid = preciseShardingValue.getValue();String resultTableName = pertab + ((uid + 1) % (dbCount * tabCount) / tabCount + 1);if (collection.contains(resultTableName)){return resultTableName;}throw new UnsupportedOperationException("route: " + resultTableName + " is not supported, please check your config");}/*** 范围分片规则(对应的是where ??>='XXX' and ??<='XXX')* 范围查询分片算法(分片键涉及区间查询时会进入该方法进行分片计算)*/@Overridepublic Collection<String> doSharding(Collection<String> collection, RangeShardingValue<Long> rangeShardingValue) {List<String> result = new ArrayList<>();Range<Long> valueRange = rangeShardingValue.getValueRange();Long upperEndpoint = valueRange.upperEndpoint();Long aLong = valueRange.lowerEndpoint();// TODO 进行相应的分片判断
// return result;return collection;}private String getPertab(Properties props) {return props.getProperty(PERTAB);}private Integer getDbCount(Properties props) {String count = props.getProperty(DB_COUNT);return count == null || count.isEmpty() ? 0 : Integer.valueOf(count);}private Integer getTabCount(Properties props) {String count = props.getProperty(TAB_COUNT);return count == null || count.isEmpty() ? 0 : Integer.valueOf(count);}@Overridepublic Properties getProps() {return props;}@Overridepublic String getType() {return "HS";}
}
需要添加一个SPI的配置文件org.apache.shardingsphere.sharding.spi.ShardingAlgorithm
,在该文件中指定我们上方创建的java类
yml配置文件中进行相应的更改
踩的坑
我先是自定义的类实现的是ComplexKeysShardingAlgorithm接口,但是我们yml配置类中还是一直按照standard
的配置,导致我自定义的类中的doSharding()
方法所以就一直没有调用到
之后我修改了complex就能调用了
在配置还是standard时,我通过debug,发现init()
和 getType()
方法都能够调用,证明SPI机制相关的文件没问题。
我就想会不会是单分片键、精确查询、范围查询相关问题导致的?
我修改了实现接口,改为了StandardShardingAlgorithm,然后就进入了其中单分片键的doSharding()
方法。最后就一点一点的排查,再到了这上面的配置