雪花算法分配DataCenterId和WorkerId的一种思路
2023-04-29 14:31:11 889
雪花算法是其中一个用于解决分布式 id 的高效方案
因其具有自增的特性, 所以更符合b+tree的索引结构
SnowFlake 算法的优点：
- 高性能高可用：生成时不依赖于数据库，完全在内存中生成
- 高吞吐：每秒钟能生成数百万的自增 ID
- ID 自增：存入数据库中，索引效率高
SnowFlake 算法的缺点：
- 依赖与系统时间的一致性，如果系统时间被回调，或者改变，可能会造成 ID 冲突或者重复
- 其中10bit-工作机器id，如果手动设置重复也可能会造成 ID 冲突或者重复
在mybatis-plus中, 没有设置机器id时，会通过当前物理网卡地址和jvm的进程id自动生成。一般在一个集群中，MAC+JVM进程PID一样的几率非常小, 但是小并不是不可能.
mybatis-plus的 issue 有很多反馈了id重复的问题. 基本都是同一机器内通过docker部署多实例导致机器id重复, 进而导致id重复
这里分享一种思路
package com.thy.backend.parent.framework.mp;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.incrementer.DefaultIdentifierGenerator;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.redisson.config.SingleServerConfig;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.SmartLifecycle;
import org.springframework.context.annotation.Bean;
import java.util.Objects;
import java.util.Random;
import java.util.concurrent.TimeUnit;
/**
* <p>SnowflakeRedissonConfig</p>
*
* @author zzx
* @version 1.0
* @date 2023/4/26 18:18:46
*/
@Slf4j
@Data
@ConfigurationProperties(prefix = "snowflake.redisson")
public class SnowflakeRedissonConfig implements SmartLifecycle {
private String address;
private Integer database;
private String password;
private static long WORKER_ID;
private static long DATA_CENTER_ID;
private static String KEY;
private static RLock LOCK;
public static final int MAX_RETRY = 10;
private volatile boolean running = false;
@Bean
public DefaultIdentifierGenerator defaultIdentifierGenerator(SnowflakeRedissonConfig redissonConfig)
throws Exception {
Config config = new Config();
SingleServerConfig singleConfig = config.useSingleServer();
singleConfig.setAddress(redissonConfig.getAddress());
singleConfig.setDatabase(redissonConfig.getDatabase());
if (StrUtil.isNotBlank(redissonConfig.getPassword())) {
config.useSingleServer().setPassword(redissonConfig.getPassword());
}
singleConfig.setConnectionPoolSize(1);
singleConfig.setConnectionMinimumIdleSize(1);
singleConfig.setSubscriptionConnectionMinimumIdleSize(1);
singleConfig.setSubscriptionConnectionPoolSize(1);
RedissonClient client = Redisson.create(config);
Random random = new Random();
try {
int inc = 0;
while (true) {
WORKER_ID = random.nextLong(1, 31);
DATA_CENTER_ID = random.nextLong(1, 31);
KEY = "snowflake_lock_" + DATA_CENTER_ID + "_" + WORKER_ID;
LOCK = client.getFairLock(KEY);
// wait 3s
boolean isLock = LOCK.tryLock(3, TimeUnit.SECONDS);
if (isLock) {
break;
}
if (inc >= MAX_RETRY) {
throw new Exception("雪花ID数据节点冲突: " + KEY);
}
inc++;
}
log.info("current snowflake node: {}", KEY);
} catch (Exception e) {
log.error("defaultIdentifierGenerator, lockKey: {}", KEY, e);
throw e;
}
return new DefaultIdentifierGenerator(WORKER_ID, DATA_CENTER_ID);
}
public static long getWorkerId() {
return WORKER_ID;
}
public static long getDataCenterId() {
return DATA_CENTER_ID;
}
@Override
public int getPhase() {
//在 WebServerGracefulShutdownLifecycle 那一组之后
return SmartLifecycle.DEFAULT_PHASE - 1;
}
@Override
public void start() {
this.running = true;
}
@Override
public void stop() {
this.running = false;
try {
if (Objects.nonNull(LOCK) && LOCK.isLocked()) {
LOCK.unlock();
}
} catch (Exception e) {
try {
if (Objects.nonNull(LOCK) && LOCK.isLocked()) {
LOCK.forceUnlock();
}
} catch (Exception ex) {
log.error("雪花ID数据节点释放失败, key: {}", KEY, e);
}
}
}
@Override
public boolean isRunning() {
return running;
}
}
通过生成两个1-31的随机数 (31)10=(11111)2
31二进制为五位 两个正好就是10bit的机器id
通过Redisson对其加锁, 再使用WatchDog机制进行锁的保持, 从而保证当前DataCenterId和WorderId不会重复
这个类注册了一个Lifecycle事件, 当应用被关闭时, 将锁释放
先用unlock尝试, unlock失败(FairLock会验证操作解锁的线程id是否与加锁线程的id一致, 如不一致则抛出异常)再使用forceUnlock.
forceUnlock再失败则只需要等待lock过期即可, 默认为30s
番外
看看mybatis-plus中对雪花算法 时钟回拨问题的处理方式
com.baomidou.mybatisplus.core.toolkit.Sequence#nextId
public synchronized long nextId() {
long timestamp = timeGen();
// 如果当前时间戳小于上次生成id的时间 则发生了时钟回拨
if (timestamp < lastTimestamp) {
long offset = lastTimestamp - timestamp;
if (offset <= 5) {
try {
// 若回拨小于5ms 则等待 5 << 1, 就是10ms, 注意存在cpu时间片这个概念, 并不一定是10ms
wait(offset << 1);
// 再生成一次 如果仍然小于上次的时间戳, 则抛出异常
timestamp = timeGen();
if (timestamp < lastTimestamp) {
throw new RuntimeException(String.format("Clock moved backwards. Refusing to generate id for %d milliseconds", offset));
}
} catch (Exception e) {
throw new RuntimeException(e);
}
} else {
throw new RuntimeException(String.format("Clock moved backwards. Refusing to generate id for %d milliseconds", offset));
}
}
...
}
