文章目录
1. 前言
上一篇文章：【分布式 ID】生成唯一 ID 的几种方式。前一篇文章我们介绍了分布式 ID 生成的几种方式，这篇文章就来看下美团开源项目 Leaf 是如何生成 ID 的。Leaf 这个名字是来自德国哲学家、数学家莱布尼茨的一句话： There are no two identical leaves in the world 也就是 “世界上没有两片相同的树叶”，首先如果大家想要详细了解一些设计思路，可以去看美团官方的文章：Leaf——美团点评分布式ID生成系统，项目地址：Meituan-Dianping/Leaf。
2. 项目启动示例 - MYSQL 和 Zookeepr
2.1 Leaf-segment 模式
首先我们用 git clone 把项目拉下来，然后按照项目下的文档 README_CN.md 里面的介绍开始启动，Leaf 有两种模式，分别是数据库号段和雪花算法，首先我们来看下数据库号段模式，如果你的 MYSQL 版本是 8.0 以上，pom 文件中需要修改下 mysql 连接和 druid 的版本。
然后在 leaf.properties 中配置好 MYSQL 的地址，在项目启动之前，先提前在数据库中创建好号段表。
CREATE DATABASE leaf
CREATE TABLE `leaf_alloc` (
`biz_tag` varchar(128) NOT NULL DEFAULT '',
`max_id` bigint(20) NOT NULL DEFAULT '1',
`step` int(11) NOT NULL,
`description` varchar(256) DEFAULT NULL,
`update_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`biz_tag`)
) ENGINE=InnoDB;
insert into leaf_alloc(biz_tag, max_id, step, description) values('leaf-segment-test', 1, 2000, 'Test leaf Segment Mode Get Id')
之后启动就可以了，然后可以通过接口调用。这里我们就直接用代码里面的测试类来测试就行。
如果要启动 test 里面的测试类，注意如果用的是 IDGenServiceTest，则需要修改测试包下面的 leaf.properties，同时 pom 里面的 mysql 版本也改下，改成 8.0 版本的，然后就可以启动了，如果你用的是上面我圈出来的这个 Test，就不需要改了，因为这个用的是 SpringBoot 启动，下面的 leaf-server 的配置如果之前就改好了，这里就不用再改这个 test 包下面的 leaf.properties，启动结果如下。
这个是数据库里面的数据。
2.2 Leaf-snowflake 模式 - 单节点
上面是数据库号段模式，下面是 Leaf-snowflake 方案，简单来说就是使用雪花算法 + Zookeeper 来生成 ID，Zookeeper 用来生成机器 ID，Leaf 集群启动的时候机器会向 Zookeeper 注册当前的信息，包括 ip、port、上报的时间，然后第一次注册会在 zookeeper 的根目录下生成一个 ip:port-WorkerID 的路径，由于 zookeeper 可以自动递增生成，所以不需要手动去配置每一台 Leaf 的 WorkerID，直接用 zookeeper 生成的递增 id 标识当前机器。
要启动 Leaf-snowflake 模式就比较简单了，首先你得确保启动了 zookeeper 服务，然后如果是 idea 启动的，需要配置下 VM 参数：-Djava.io.tmpdir=D:\javaCode\Leaf。配置这个参数是因为 Leaf 除了会向 zookeeper 注册信息之外，获取到了 WorkerID 之后还会把这个 WorkerID 存一份数据到本地，避免 zookeeper 出问题的时候没办法获取到 WorkerID。按官方的说法就是弱依赖 ZooKeeper。
这里我就直接用 leaf-server 服务来演示，还是一样，需要修改配置文件 leaf-properties，在里面配置上你的 zk 地址和端口，把 leaf.snowflake.enable 设置为 true。
然后启动 LeafServerApplication。
启动成功之后使用接口工具请求发起请求，或者直接把这个链接在浏览器请求。
然后打开 zookeeper，就可以发现节点已经注册上去了，/snowflake 下面的 forever 节点下面的 ip:port-workerID，可以看到我们注册上去的节点 ID 为 0。
2.3 Leaf-snowflake 模式 - 多节点
那我们要怎么模拟一个集群呢？我们可以在 IDEA 里面启动多个 LeafServerApplication，但是在这之前，我们要修改下配置。
首先，这里的配置 leaf.snowflake.port 改一下，我们把这个配置给注释掉，因为这个 port 不是用来连接 zookeeper 的，只是用来标记当前 Leaf 的启动端口，由于我们要在 IDEA 启动多个 LeafServerApplication，这里的 port 肯定不能写死。
接下来我们修改下源码，主要就是修改 SnowflakeService 的构造器，原本是直接从配置文件里面获取 port， 现在我们改成如果配置文件里面获取不到，就从系统变量中获取 server.port，也就是 Spring 项目的启动端口，接下来我们构建两个启动 Application 服务。
接下来我们就可以启动两个项目了。
然后到 zookeeper 下面看下有没有注册上去，可以看到 8080 和 8081 都注册上去了，且后面的 workerID 是递增的。
上面我们也说了，项目启动的时候不单单会注册到 Zookeeper，也会在本地存储一个 WorkerID 文件，防止 Zookeeper 出问题的时候获取不到 WorkerID，也就是上面 -Djava.io.tmpdir 这个路径。
可以看到本地存储确实也没什么问题，到这里我们就演示完了。不过大家会不会好奇，就是 zookeeper 下面这玩意是顺序递增的，那超过 1024 台 Leaf 就注册不下了？其实可以看到这个路径是 /snowflake/${leaf.name}/forever，leaf.name 是配置文件里面可以配置的，所以不是说固定存到某个路径，每一个 Leaf 都可以配置自己的 leaf.name。
3. Leaf-segment 详细讲解
这种模式就是数据库号段模式，区别于数据库自增字段，这种模式下每一次都会从数据库获取一个号段，这个号段长度是业务自己设置，接着分配 ID 就在内存中分配这些号段 ID，而不是每一次都需要请求数据库。
上面是官方的图，可以看到，biz_tag 就是用来区分业务的，max_id 指当前号段下能够分配的最大值 + 1，step 代表号段长度，比如 1000，那么分配的时候就从 1 开始分配，分配到 1000。
上面图中 test_tag 在不同机器上面的号段不同，这时候如果第一台机器的 1~ 1000 分配完了，再次去数据库获取号段，就会获取到 3001~4000 的号段。这种方式可以将请求数据库的频率减少到 1/step，每一次请求数据库都是在分配的 step 长度的号段被消耗完之后才去请求数据库再次分配，大大提高了性能。
这种方式有一定的容灾性，在 DB 宕机的时候由于本地还有号段，所以还可以继续提供服务，当然了这种强依赖数据库的方式一旦 DB 宕机，服务不可用也是时间问题，而且每一次消耗完号段之后还得立刻去请求数据库，这样一来获取 ID 的请求就会阻塞住，一旦请求数据库有延迟，那么业务也会收到影响，所以为了解决这个问题，Leaf 设计了双 Buffer 的方式来解决这个问题。
设计了双 Buffer 之后，当当前号段消费到 10% 的时候，如果下一个号段还没有准备好并且更新号段的线程没有在执行，那么立刻使用线程去更新下一个号段，由于更新是使用了线程去执行，所以当前业务是不受影响的。
而且这里的设计是当当前号段消费到 10% 就立刻去请求处理下一个号段，当然这个比例我们没办法得知为什么要设置到 10%，如果可以设置成动态配置那就更好了，不过我们也可以发现如果这个比例设置的太高，假设设置到了 90%，那么会导致如果某一个时刻有大批量请求到来瞬间把这 10% 的号段消耗了，那么剩下的请求又要阻塞等待，所以设置的小一点是没问题的。
那么 segment 长度设置成多少合适呢？官方推荐 segment 长度设置为服务高峰期发号 QPS 的 600倍（10分钟），这样即使 DB 宕机，Leaf 仍能持续发号 10-20 分钟不受影响。实际上源码中也会根据请求数据库的频率来对 segment 进行减半或者是加倍。
4. Leaf-segment 源码解析
Leaf-segment 的源码主要集中在 leaf-core 下面的 SegmentIDGenImpl。
先来看下号段模式下的一些对象。
4.1 SegmentBuffer 号段缓存
SegmentBuffer 就是号段 Buffer 缓存，里面存储了双 Segment 交替来使用。
上面是大致的结构，下面来看下里面的属性。
/**
* 双 buffer
*/
public class SegmentBuffer {
private String key;
private Segment[] segments; // 双 buffer, 交换来使用
private volatile int currentPos; // 当前的使用的 segment 的 index
private volatile boolean nextReady; //下一个 segment 是否处于可切换状态
private volatile boolean initOk; // 是否初始化完成
private final AtomicBoolean threadRunning; // 线程是否在运行中
private final ReadWriteLock lock;
// 步长
private volatile int step;
// 最小步长
private volatile int minStep;
// 更新时间
private volatile long updateTimestamp;
...
}
其中 threadRunning 表示第二个 segment 是不是在线程中执行准备逻辑，然后 step 就是步长，由于步长是可以根据请求数据库的时间来扩容的和缩容的，所以有一个 minStep 代表最小步长，也就是说这个步长就算再怎么缩小也不会小于这个数，最后一个 updateTimestamp 就是当请求数据库获取号段的时候会更新。
4.2 Segment 号段
public class Segment {
// 当前要分配的 ID 值
private AtomicLong value = new AtomicLong(0);
// 这个号段可以分配的 ID 的最大值 + 1
private volatile long max;
// 号段分配的步长, 也是号段的大小
private volatile int step;
// 这个段所属的 buffer
private SegmentBuffer buffer;
...
}
Segment 的注释写的比较清楚了，下面是里面的结构，注意一下 max 是这个号段下可以分配的最大值 + 1，比如这个号段 step 是 1000，value 是 2001，那么当前号段可分配的范围就是 [2001，3000]，共 1000 个数（step）， max 就是 3001。
4.3 初始化号段服务 SegmentIDGenImpl
回到 SegmentIDGenImpl 方法，里面的 init 方法是在创建 SegmentService 的时候创建出来的，SegmentService 提供了方法，比如 getId，getIdGen，里面的实现则是在 SegmentIDGenImpl 中完成。
public SegmentService() throws SQLException, InitException {
Properties properties = PropertyFactory.getProperties();
// 准备 MYSQL
boolean flag = Boolean.parseBoolean(properties.getProperty(Constants.LEAF_SEGMENT_ENABLE, "true"));
if (flag) {
// Config dataSource
dataSource = new DruidDataSource();
dataSource.setUrl(properties.getProperty(Constants.LEAF_JDBC_URL));
dataSource.setUsername(properties.getProperty(Constants.LEAF_JDBC_USERNAME));
dataSource.setPassword(properties.getProperty(Constants.LEAF_JDBC_PASSWORD));
dataSource.init();
// 准备 DAO
IDAllocDao dao = new IDAllocDaoImpl(dataSource);
// 创建 SegmentIDGenImpl
idGen = new SegmentIDGenImpl();
((SegmentIDGenImpl) idGen).setDao(dao);
// 初始化
if (idGen.init()) {
logger.info("Segment Service Init Successfully");
} else {
throw new InitException("Segment Service Init Fail");
}
} else {
// 如果 leaf.segment.enable 这个配置没有设置为 true, 那么 ID 生成器的实现就是 ZeroIDGen, ID 直接返回 0
idGen = new ZeroIDGen();
logger.info("Zero ID Gen Service Init Successfully");
}
}
SegmentService 的创建主要是完成两部分操作：
- 准备数据库配置。
- 创建 SegmentIDGenImpl，前提是
leaf.segment.enable这个配置设置为 true，代表使用号段模式，如果没有设置，那么创建的 ID 生成器就是ZeroIDGen，这个生成器是统一返回 0 的。
而 idGen.init() 方法就是核心的逻辑，在里面会去初始化 SegmentIDGenImpl，下面来看下源码。
@Override
public boolean init() {
logger.info("Init ...");
// 确保加载到 kv 后才初始化成功
updateCacheFromDb();
initOK = true;
// 初始化定时任务, 启动后 60s 开始执行, 之后每隔 1min 执行一次
updateCacheFromDbAtEveryMinute();
return initOK;
}
private void updateCacheFromDbAtEveryMinute() {
// 创建检测缓存的任务线程池
ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(r);
t.setName("check-idCache-thread");
t.setDaemon(true);
return t;
}
});
service.scheduleWithFixedDelay(new Runnable() {
@Override
public void run() {
// 定时更新数据库的 tag 到缓存中
updateCacheFromDb();
}
}, 60, 60, TimeUnit.SECONDS);
}
初始化主要完成两个操作，首先是从 DB 中拉取 biz_tag 来为这些业务创建 SegmentBuffer，然后启动一个定时任务，启动后 60s 开始执行, 之后每隔 1min 执行一次，同样调用 updateCacheFromDb 方法去维护新增的 tag，所以根据这一段介绍就可以明显得知了 updateCacheFromDb 就是去维护数据库的 biz_tag 的。
上面的两个步骤就是这个方法要做的，主要就是同步数据库和缓存的一致性，注意这里 init 之后 SegmentBuffer 里面的 initOk 还是 false，相当于说这里只是同步了一下数据库和缓存，创建缓存中不存在的 SegmentBuffer，同时将数据库中已经删掉的 SegmentBuffer，缓存也删掉。
// 从数据库拉取所有业务的号段下来
// 1. 初始化调用
// 2. 定时任务调用
private void updateCacheFromDb() {
logger.info("update cache from db");
StopWatch sw = new Slf4JStopWatch();
try {
// 首先拉取所有的业务 tag
List<String> dbTags = dao.getAllTags();
if (dbTags == null || dbTags.isEmpty()) {
return;
}
// 缓存里面现有的 tag
List<String> cacheTags = new ArrayList<String>(cache.keySet());
// 需要新增的 tag, 也就是数据库有但是缓存里面没有的
Set<String> insertTagsSet = new HashSet<>(dbTags);
// 需要删除的 tag, 也就是数据库没有但是缓存有的
Set<String> removeTagsSet = new HashSet<>(cacheTags);
// 遍历缓存里面的 tag, 这里是定时任务遍历的时候才不会为空了
for(int i = 0; i < cacheTags.size(); i++){
String tmp = cacheTags.get(i);
// 如果缓存里面的 tag 在数据库中出现了, 就删掉, 剩下的就是缓存里面没有的
// 比如缓存里面是 tag1、tag2、tag3, 数据库里面是 tag2、tag3、tag4
// 遍历完成之后 insertTagsSet 里面就只剩下 tag4 了
if(insertTagsSet.contains(tmp)){
insertTagsSet.remove(tmp);
}
}
// 遍历需要新增的 tag
for (String tag : insertTagsSet) {
// 新建一个 SegmentBuffer 对象, 一个 buffer 里面有两个 segment 交替来使用, 注意这里只是新建 SegmentBuffer 和
// Segment, 不是说初始化完成了就可以立马开始使用里面的号段
SegmentBuffer buffer = new SegmentBuffer();
buffer.setKey(tag);
// 创建第一个号段
Segment segment = buffer.getCurrent();
// 当前要分配的值, 初始化为 0
segment.setValue(new AtomicLong(0));
// 当前号段能分配的最大值 + 1
segment.setMax(0);
// 分配步长
segment.setStep(0);
// 添加到缓存中
cache.put(tag, buffer);
logger.info("Add tag {} from db to IdCache, SegmentBuffer {}", tag, buffer);
}
// 将缓存中已经失效的 tag 删掉, 遍历数据库的 tag
for(int i = 0; i < dbTags.size(); i++){
String tmp = dbTags.get(i);
// 如果缓存里面 tag 还在数据库中, 就从集合中删掉
// 比如缓存里面是 tag1、tag2、tag3, 数据库里面是 tag2、tag3、tag4
// 遍历完成之后 removeTagsSet 就剩下 tag1 了
if(removeTagsSet.contains(tmp)){
removeTagsSet.remove(tmp);
}
}
// 将失效的 tag 删掉
for (String tag : removeTagsSet) {
cache.remove(tag);
logger.info("Remove tag {} from IdCache", tag);
}
} catch (Exception e) {
logger.warn("update cache from db exception", e);
} finally {
sw.stop("updateCacheFromDb");
}
}
可以看到在处理 insertTagsSet 的时候，创建出了 buffer 之后只是设置了一个 key，而构造里里面也除了创建出两个 Segment，其他属性都没有设置。
public SegmentBuffer() {
segments = new Segment[]{new Segment(this), new Segment(this)};
currentPos = 0;
nextReady = false;
initOk = false;
threadRunning = new AtomicBoolean(false);
lock = new ReentrantReadWriteLock();
}
4.4 核心方法 get 获取 ID
下面就是核心方法，如何通过 get 去获取号段 ID。
/**
* 获取 id
* @param key 业务 key
* @return
*/
@Override
public Result get(final String key) {
if (!initOK) {
// 还没初始化好
return new Result(EXCEPTION_ID_IDCACHE_INIT_FALSE, Status.EXCEPTION);
}
// 如果缓存里面有这个 key, 就是数据库里面的 biz_tag
if (cache.containsKey(key)) {
// 获取 SegmentBuffer
SegmentBuffer buffer = cache.get(key);
// 双重检查锁
if (!buffer.isInitOk()) {
synchronized (buffer) {
if (!buffer.isInitOk()) {
try {
// 如果 buffer 还没有初始化, 就初始化, 注意这里 updateCacheFromDb 方法中通过 new
// 创建出来的 SegmentBuffer 是还没有初始化的, 会在这里去初始化
updateSegmentFromDb(key, buffer.getCurrent());
logger.info("Init buffer. Update leafkey {} {} from db", key, buffer.getCurrent());
buffer.setInitOk(true);
} catch (Exception e) {
logger.warn("Init buffer {} exception", buffer.getCurrent(), e);
}
}
}
}
// 从号段中获取 id
return getIdFromSegmentBuffer(cache.get(key));
}
// 没找到这个业务有号段信息
return new Result(EXCEPTION_ID_KEY_NOT_EXISTS, Status.EXCEPTION);
}
首先如果 cache 缓存里面没找到这个 tag，说明这个业务没有设置到数据库，或者设置到数据库了但是还没有同步到缓存中。
而对于刚同步过来的 SegmentBuffer，由于 initOk 是 false，所以会通过 updateSegmentFromDb 真正初始化 SegmentBuffer，这里真正初始化的意思就是读取数据里面的号段信息设置到 SegmentBuffer 中。
最后再通过 getIdFromSegmentBuffer 从号段中获取下一个 ID 返回。
4.4.1 updateSegmentFromDb 更新 Segment
/**
* 从数据库获取这个 tag 的信息来更新号段
* @param key
* @param segment
*/
public void updateSegmentFromDb(String key, Segment segment) {
StopWatch sw = new Slf4JStopWatch();
// 获取这个号段所属的 SegmentBuffer
SegmentBuffer buffer = segment.getBuffer();
LeafAlloc leafAlloc;
// 如果还没有初始化
if (!buffer.isInitOk()) {
// 通过业务 tag 查询到号段信息
leafAlloc = dao.updateMaxIdAndGetLeafAlloc(key);
// 设置号段长度
buffer.setStep(leafAlloc.getStep());
// 设置最小号段长度
buffer.setMinStep(leafAlloc.getStep());
} else if (buffer.getUpdateTimestamp() == 0) {
// 第一次更新, 比如前一个号段使用已经超过 10% 了, 就会去提前更新下一个号段的信息
leafAlloc = dao.updateMaxIdAndGetLeafAlloc(key);
// 设置更新时间是当前时间
buffer.setUpdateTimestamp(System.currentTimeMillis());
// 设置号段长度
buffer.setStep(leafAlloc.getStep());
// 设置号段的最小长度
buffer.setMinStep(leafAlloc.getStep());
} else {
// 距离上一次更新的时间有多久
long duration = System.currentTimeMillis() - buffer.getUpdateTimestamp();
// 当前 buffer 的步长
int nextStep = buffer.getStep();
// 如果更新时间小于号段的过期时间, 说明消费的有点快
if (duration < SEGMENT_DURATION) {
// 号段扩容不能超过 1000000
if (nextStep * 2 > MAX_STEP) {
//do nothing
} else {
// 扩容成 2 倍
nextStep = nextStep * 2;
}
} else if (duration < SEGMENT_DURATION * 2) {
// 这里是更新时间超过了号段过期时间, 说明当前的号段长度够用了
} else {
// 这里就是消费得太慢了, 缩容成原来的一半, 这里的最小步长是一开始创建的时候设置的, 后面不会再更新了
nextStep = nextStep / 2 >= buffer.getMinStep() ? nextStep / 2 : nextStep;
}
logger.info("leafKey[{}], step[{}], duration[{}mins], nextStep[{}]", key, buffer.getStep(), String.format("%.2f",((double)duration / (1000 * 60))), nextStep);
// 更新数据库的号段信息
LeafAlloc temp = new LeafAlloc();
temp.setKey(key);
temp.setStep(nextStep);
leafAlloc = dao.updateMaxIdByCustomStepAndGetLeafAlloc(temp);
// 设置当前 buffer 的更新时间、号段大小和最小号段
buffer.setUpdateTimestamp(System.currentTimeMillis());
buffer.setStep(nextStep);
buffer.setMinStep(leafAlloc.getStep());//leafAlloc的step为DB中的step
}
// 设置下一次要分配的 id 值, 上面在 updateMaxIdByCustomStepAndGetLeafAlloc 之后 maxId 已经更新成 maxId + step 了
// 比如 step 原来是 2000, 现在扩容成 4000 之后, maxId 也会被更新成 maxId + step, maxId 在创建数据库的时候默认是 1,
// 也就是说现在扩容之后 maxId 就变成了 4001, value = 2001
long value = leafAlloc.getMaxId() - buffer.getStep();
segment.getValue().set(value);
segment.setMax(leafAlloc.getMaxId());
segment.setStep(buffer.getStep());
sw.stop("updateSegmentFromDb", key + " " + segment);
}
这个方法就是从数据库获取这个 tag 的信息来更新号段，主要分为三个阶段，每个阶段有不同的逻辑。
- 还没有初始化的时候。
- 初始化了，但是是第一次更新信息，比如前一个号段使用已经超过 10% 了, 就会去提前更新下一个号段的信息。
- 已经更新很多次了。
首先是还没有初始化的时候，也就是 !buffer.isInitOk()，会去调用 updateMaxIdAndGetLeafAlloc 更新 max_id，我们之前说过 max_id 在创建数据表的时候就已经指定了默认是 1，而 updateMaxIdAndGetLeafAlloc 的逻辑就是先 updateMaxId 再 getLeafAlloc，就是下面的两个 SQL。
所以假设设置了 step = 2000，那么 max_id 会被更新成 2001。当获取到 leafAlloc 后，设置号段长度和最小号段长度为数据库的 step，然后初始化完成了。
然后假设是已经初始化过了，但是当前是第一次更新，可以看到上面初始化的逻辑里面没有设置 buffer 的 updateTimestamp，所以这一次更新会进入 else if (buffer.getUpdateTimestamp() == 0) 这个分支，逻辑跟上面的都一样，只是多了一个 buffer.setUpdateTimestamp(System.currentTimeMillis()) 设置更新时间。
最后是已经初始化过并且不是第一次更新了，就会进入最后的 else 的逻辑。首先会获取当前距离上一次更新的时间有多久（duration），然后获取当前 buffer 的步长 nextStep，接下来根据更新的间隔时间去判断步长是否需要扩大还是缩小。
- 如果说更新的时间小于号段时间的过期时间（15 分钟），那么说明消费有点快了，这种情况下将号段扩容成原来的 2 倍，但是最大不能超过 1000000。
- 如果距离上一次更新的时间超过了号段过期时间但是小于两倍的号段过期时间，说明现在这个长度适合业务的消费速度，不需要修改。
- 如果说距离上一次更新时间超过了号段过期时间的 2 倍，说明这个号段长度太长了，有可能是扩容之后太大了，业务不需要消费这么快，这时候缩容成原来的 1/2。
为什么不将号段设置的很大，这样就不需要扩容了呢？ 下面是我的理解，不一定对，因为可能有多台 Leaf 部署来承担多个业务的 ID 生成，集群可能很大，如果号段设置得很大，就会导致每一台机器都分配到一个很大的号段，但是业务消费速度慢，这种情况下如果集群有问题，重新启动又得重新分配号段，就会比较浪费，当然这是我的理解，如果还有其他方面的问题也可以讨论。
回到源码，最后计算下 value 值，也就是下一次要分配的 id 值，上面在 updateMaxIdByCustomStepAndGetLeafAlloc 之后 maxId 已经更新成 maxId + step 了，比如 step 原来是 2000，现在扩容成 4000 之后，maxId 也会被更新成 maxId + step，maxId 在创建数据库的时候默认是 1，也就是说现在扩容之后 maxId 就变成了 4001，value = 2001，本次从 2001 开始继续分配。然后设置 Segment 的 value、max、step，结束。
4.4.2 getIdFromSegmentBuffer 从号段中获取 ID
当更新完 Segment 后，调用 getIdFromSegmentBuffer 从号段中获取 ID，下面是全部逻辑。
/**
* 从号段中获取 ID
* @param buffer
* @return
*/
public Result getIdFromSegmentBuffer(final SegmentBuffer buffer) {
// 死循环获取
while (true) {
// 加读锁, 防止业务并发对同一个号段获取
buffer.rLock().lock();
try {
// 获取当前正在使用的号段
final Segment segment = buffer.getCurrent();
// 如果当前的 buffer 下一个号段还没有准备好, 同时当前号段还剩下的 ID 数小于 90%, 开始准备下一个号段
if (!buffer.isNextReady() && (segment.getIdle() < 0.9 * segment.getStep()) && buffer.getThreadRunning().compareAndSet(false, true)) {
service.execute(new Runnable() {
@Override
public void run() {
// 获取下一个要使用的号段, SegmentBuffer 是两个 buffer 交替使用
Segment next = buffer.getSegments()[buffer.nextPos()];
boolean updateOk = false;
try {
// 从数据库获取下一个号段信息来更新 Segment
updateSegmentFromDb(buffer.getKey(), next);
updateOk = true;
logger.info("update segment {} from db {}", buffer.getKey(), next);
} catch (Exception e) {
logger.warn(buffer.getKey() + " updateSegmentFromDb exception", e);
} finally {
// 更新下一个号段成功
if (updateOk) {
// 加写锁
buffer.wLock().lock();
// 设置属性
buffer.setNextReady(true);
buffer.getThreadRunning().set(false);
// 解除写锁
buffer.wLock().unlock();
} else {
// 解除失败
buffer.getThreadRunning().set(false);
}
}
}
});
}
// 上面更新下一个号段成功后, 获取当前要分配的 ID
long value = segment.getValue().getAndIncrement();
// 返回结果
if (value < segment.getMax()) {
return new Result(value, Status.SUCCESS);
}
} finally {
// 解除读锁
buffer.rLock().unlock();
}
// 这里就是获取号段失败, 比如当前号段已经 >= segment.getMax, 在里面会去阻塞等到下一个号段准备成功
waitAndSleep(buffer);
// 加写锁
buffer.wLock().lock();
try {
// 获取当前的号段
final Segment segment = buffer.getCurrent();
// 再次判断如果当前的号段是符合要求的, 就返回，可能是因为这里两个线程同时来获取，然后当前线程上下文切换给其他线程，其他线程先一步完成了 switchPos，那么当当前线程再获取 buffer.getCurrent() 的时候获取到的就是有值的号段
long value = segment.getValue().getAndIncrement();
if (value < segment.getMax()) {
return new Result(value, Status.SUCCESS);
}
// 到这里当前号段用完了, 判断下一个是否准备好了
if (buffer.isNextReady()) {
// 切换到下一个 Segment
buffer.switchPos();
// 设置下一个 Segment 的 nextReady 为 false
buffer.setNextReady(false);
} else {
logger.error("Both two segments in {} are not ready!", buffer);
return new Result(EXCEPTION_ID_TWO_SEGMENTS_ARE_NULL, Status.EXCEPTION);
}
} finally {
// 解除写锁
buffer.wLock().unlock();
}
}
}
获取 ID 就是在一个 while 循环里面去获取的，首先加读锁，然后重点来了，判断如果当前的 buffer 下一个号段还没有准备好，同时当前号段还剩下的 ID 数小于 90%，开始准备下一个号段，所谓的准备下一个号段就是提前给下一个号段分配好值，分配的方法就是上面的 updateSegmentFromDb，当分配好之后当前号段如果用完了，就可以无缝切换到下一个号段。当线程中更新完号段了， nextReady 就会设置为 true，同时将 threadRunning 设置为 false，防止并发更新下一个号段。
上面是更新号段的流程，如果说当前号段分配的 value 比最大值要小，就可以直接返回结果，但是如果不是，就说明当前号段用完了，这种情况下需要阻塞等到下一个号段准备成功。上面我们也说了如果下一个号段准备成功 threadRunning 会设置为 false，所以 waitAndSleep 就是在 while 循环里面不断判断这个标记。
private void waitAndSleep(SegmentBuffer buffer) {
int roll = 0;
// 一直死循环阻塞等待下一个号段准备成功
while (buffer.getThreadRunning().get()) {
roll += 1;
if(roll > 10000) {
try {
TimeUnit.MILLISECONDS.sleep(10);
break;
} catch (InterruptedException e) {
logger.warn("Thread {} Interrupted",Thread.currentThread().getName());
break;
}
}
}
}
当然了准备好后，继续判断当前号段分配的值是否是符合要求的，再次判断猜测是因为如果当前的号段是符合要求的, 就返回，可能是因为这里两个线程同时来获取，然后当前线程上下文切换给其他线程，其他线程先一步完成了 switchPos，那么当当前线程再获取 buffer.getCurrent() 的时候获取到的就是有值的号段，因为用的是读写锁，并非单一的锁？
如果还是不符合，就切换到下一个号段，然后设置 nextReady 为 false，下一次获取 ID 就继续去提前分配号段，最后解除写锁。
5. Leaf-snowflake 源码解析
上面是数据库号段模式，下面是 Leaf-snowflake，Leaf-snowflake 相比号段模式就简单不少了，因为不设计数据库的读取，逻辑上比较简单，核心就是上一篇文章的雪花算法，而 Leaf 在节点的持久化做了不少工作，下面来看下这个模式的源码。
5.1 SnowflakeIDGenImpl 的初始化
snowflake 模式的具体实现是 SnowflakeIDGenImpl，这个类的 init 方法是会返回 true 的，所以不用管。
@Override
public boolean init() {
return true;
}
我们主要看下构造器，看看这个类型的 ID 生成器创建的时候做了什么。
public SnowflakeIDGenImpl(String zkAddress, int port) {
// Thu Nov 04 2010 09:42:54 GMT+0800 (中国标准时间)
this(zkAddress, port, 1288834974657L);
}
/**
* @param zkAddress zk地址
* @param port snowflake 监听端口
* @param twepoch 起始的时间戳
*/
public SnowflakeIDGenImpl(String zkAddress, int port, long twepoch) {
// 启动时间设置为 Thu Nov 04 2010 09:42:54 GMT+0800 (中国标准时间)
this.twepoch = twepoch;
Preconditions.checkArgument(timeGen() > twepoch, "Snowflake not support twepoch gt currentTime");
// 获取当前的服务器 IP
final String ip = Utils.getIp();
// 监听端口默认是 8080
SnowflakeZookeeperHolder holder = new SnowflakeZookeeperHolder(ip, String.valueOf(port), zkAddress);
LOGGER.info("twepoch:{} ,ip:{} ,zkAddress:{} port:{}", twepoch, ip, zkAddress, port);
// 初始化
boolean initFlag = holder.init();
if (initFlag) {
workerId = holder.getWorkerID();
LOGGER.info("START SUCCESS USE ZK WORKERID-{}", workerId);
} else {
Preconditions.checkArgument(initFlag, "Snowflake Id Gen is not init ok");
}
Preconditions.checkArgument(workerId >= 0 && workerId <= maxWorkerId, "workerID must gte 0 and lte 1023");
}
首先就是基准时间，因为雪花算法需要一个基准时间，这样设置 41 位时间戳的时候就可以用当前时间 - 基准时间来设置，我们上一篇文章说过，雪花算法完整可以用 69 年，也就是可以用到 基准时间 + 69 年，Leaf 默认是 Thu Nov 04 2010 09:42:54 GMT+0800 (中国标准时间)。
接下来创建 SnowflakeZookeeperHolder，这里面就是去维护节点信息的，比如节点上报 Zookeeper 等，port 不是 Zookeeper 的端口，而是标识当前 Leaf 服务的监听端口，用于创建 Zookeeper 用的。
接下来就调用 init 去初始化，初始化完了之后获取 workerId。
5.2 SnowflakeZookeeperHolder 的初始化
private String zk_AddressNode = null;// 保存自身的key ip:port-000000001
private String listenAddress = null;// 保存自身的key ip:port
// WorkerID, 也就是上面 ip:port-000000001 后面的数字
private int workerID;
// 持久化到 Zookeeper 的哪个路径下
private static final String PREFIX_ZK_PATH = "/snowflake/" + PropertyFactory.getProperties().getProperty("leaf.name");
private static final String PROP_PATH = System.getProperty("java.io.tmpdir") + File.separator + PropertyFactory.getProperties().getProperty("leaf.name") + "/leafconf/{port}/workerID.properties";
private static final String PATH_FOREVER = PREFIX_ZK_PATH + "/forever";//保存所有数据持久的节点
// ip 地址
private String ip;
// 当前服务监听端口
private String port;
// Zookeeper 连接信息
private String connectionString;
// 上一次上报到 Zookeeper 的时间
private long lastUpdateTime;
在看 init 方法的源码之前，我们先来看下这个类里面的一些属性。
- PATH_FOREVER：持久化的 Zookeeper 路径，最终持久化的路径是
/snowflake/${leaf.name}/forever，leaf.name 可以在配置文件 leaf.properties 里面去配置。 - PROP_PATH ：持久化的本地文件路径，需要避免 Zookeeper 不可用时拿不到 workerID 的情况，因此需要将 workerID 在本地持久化一份，持久化的路径就是
${java.io.tmpdir}/${leaf.name}//leafconf/{port}/workerID.properties， 2.3 小节也有演示。 - zk_AddressNode：保存自身的 key，ip:port-000000001。
- workerID：上面 zk_AddressNode 中
-后面的 000000001 就是工作 ID。 - lastUpdateTime：上一次上报到 Zookeeper 的时间。
上面是一些比较重要的属性，下面就可以来看下初始化的源码了。
/**
* 初始化 zookeeper 配置
* @return
*/
public boolean init() {
try {
CuratorFramework curator = createWithOptions(connectionString, new RetryUntilElapsed(1000, 4), 10000, 6000);
curator.start();
Stat stat = curator.checkExists().forPath(PATH_FOREVER);
if (stat == null) {
// 不存在根节点,机器第一次启动,创建 /snowflake/ip:port-000000000,并上传数据
zk_AddressNode = createNode(curator);
// worker id 默认是0
updateLocalWorkerID(workerID);
// 定时上报本机时间给 forever 节点
ScheduledUploadData(curator, zk_AddressNode);
return true;
} else {
Map<String, Integer> nodeMap = Maps.newHashMap();//ip:port->00001
Map<String, String> realNode = Maps.newHashMap();//ip:port->(ipport-000001)
// 存在根节点,先检查是否有属于自己的根节点
List<String> keys = curator.getChildren().forPath(PATH_FOREVER);
for (String key : keys) {
String[] nodeKey = key.split("-");
realNode.put(nodeKey[0], key);
nodeMap.put(nodeKey[0], Integer.parseInt(nodeKey[1]));
}
// 是否存在当前机器的节点, 如果存在说明不是第一次上报
Integer workerid = nodeMap.get(listenAddress);
if (workerid != null) {
// 有自己的节点,zk_AddressNode=ip:port
zk_AddressNode = PATH_FOREVER + "/" + realNode.get(listenAddress);
workerID = workerid;// 启动 worder 时使用会使用
if (!checkInitTimeStamp(curator, zk_AddressNode)) {
// 判断是否发生了时间回退
throw new CheckLastTimeException("init timestamp check error,forever node timestamp gt this node time");
}
// 准备创建临时节点
doService(curator);
// 在本地节点文件系统上缓存一个 workid 值,zk 失效,机器重启时保证能够正常启动
updateLocalWorkerID(workerID);
LOGGER.info("[Old NODE]find forever node have this endpoint ip-{} port-{} workid-{} childnode and start SUCCESS", ip, port, workerID);
} else {
// 表示新启动的节点, 创建持久节点, 不用 check 时间
String newNode = createNode(curator);
zk_AddressNode = newNode;
String[] nodeKey = newNode.split("-");
// 节点 ID
workerID = Integer.parseInt(nodeKey[1]);
doService(curator);
// 在本地节点文件系统上缓存一个 workid 值,zk 失效,机器重启时保证能够正常启动
updateLocalWorkerID(workerID);
LOGGER.info("[New NODE]can not find node on forever node that endpoint ip-{} port-{} workid-{},create own node on forever node and start SUCCESS ", ip, port, workerID);
}
}
} catch (Exception e) {
LOGGER.error("Start node ERROR {}", e);
try {
// 启动异常, 从本地文件中加载 workerId
Properties properties = new Properties();
properties.load(new FileInputStream(new File(PROP_PATH.replace("{port}", port + ""))));
workerID = Integer.valueOf(properties.getProperty("workerID"));
LOGGER.warn("START FAILED ,use local node file properties workerID-{}", workerID);
} catch (Exception e1) {
LOGGER.error("Read file error ", e1);
return false;
}
}
return true;
}
可以看到首先如果 PATH_FOREVER 这个路径在 Zookeeper 中不存在，说明是第一次上报信息，这种情况下创建一个节点下来，创建节点路径是 /snowflake/${leaf.name}/forever/ip:port-，由于是顺序创建，所以会在路径后面自动拼接上编号，就是我们当前机器的 WorkerID。
/**
* 创建持久顺序节点 ,并把节点数据放入 value
*
* @param curator
* @return
* @throws Exception
*/
private String createNode(CuratorFramework curator) throws Exception {
try {
// 由于是顺序节点, 所以创建出来的路径就是 /snowflake/${leaf.name}/forever/ip:port-00000
// /snowflake/${leaf.name}/forever/ip:port-00001
// ...
return curator.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT_SEQUENTIAL).forPath(PATH_FOREVER + "/" + listenAddress + "-", buildData().getBytes());
} catch (Exception e) {
LOGGER.error("create node error msg {} ", e.getMessage());
throw e;
}
}
当然了这里的 WorkerID 第一次创建就默认是 0 来着，所以不需要再设置。updateLocalWorkerID 就是刚刚说的，Leaf 为了解决 Zookeeper 不可用时获取不到 workerID 的问题，会在初始化的时候也往本地文件 ${java.io.tmpdir}/${leaf.name}//leafconf/{port}/workerID.properties 中存储一份 workerID 的信息，最后通过 ScheduledUploadData 定时上报本机的时间给 Zookeeper。
那如果不是第一次创建，就说明可能是 Leaf 重启什么的，这种情况下先获取这个 PATH_FOREVER 下面的所有节点，可以看下面图。
获取到这些节点之后，设置到 realNode 和 nodeMap 中，然后通过当前机器的 ip+port 来获取对于的 workerID。
如果能获取到就通过 checkInitTimeStamp 判断上一次上报的时间有没有大于当前时间，如果大于说明有问题，有可能发生了时钟回拨，这种情况下抛出异常 CheckLastTimeException。
private boolean checkInitTimeStamp(CuratorFramework curator, String zk_AddressNode) throws Exception {
byte[] bytes = curator.getData().forPath(zk_AddressNode);
Endpoint endPoint = deBuildData(new String(bytes));
// 该节点的时间不能小于最后一次上报的时间
return !(endPoint.getTimestamp() > System.currentTimeMillis());
}
如果都没有问题，就通过 doService 启动一个定时任务，定时上报当前节点的信息到 Zookeeper 中。然后通过 updateLocalWorkerID 更新本地的文件，在本地节点文件系统上缓存一个 workID 值，zk 失效，机器重启时保证能够正常启动。
那如果是路径创建了，但是获取不到这个节点，说明这个 Leaf 节点也上报到了这个 PATH_FOREVER 路径，但是是新上报的，需要新建一个新的节点，逻辑和上面基本一样，不过这里就需要设置下 workerID，因为如果路径不存在，当前节点的 workerID 肯定是 0，不用想，但是如果路径存在，由于递增的特性，当前节点不一定是 0，所以创建出节点之后需要设置下 workerID，最后注意如果是新建节点是不需要 check 时间的。
最后如果说这个过程发生了异常，那么从本地文件中加载出 workerID，避免服务不可用。
5.3 doService 构建定时任务
private void doService(CuratorFramework curator) {
ScheduledUploadData(curator, zk_AddressNode);// /snowflake_forever/ip:port-000000001
}
private void ScheduledUploadData(final CuratorFramework curator, final String zk_AddressNode) {
// 创建一个单线程的线程池
Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread thread = new Thread(r, "schedule-upload-time");
thread.setDaemon(true);
return thread;
}
}).scheduleWithFixedDelay(new Runnable() {
// 定时上报节点信息到 zookeeper
@Override
public void run() {
updateNewData(curator, zk_AddressNode);
}
}, 1L, 3L, TimeUnit.SECONDS);//每3s上报数据
}
这里就是创建定时任务，初始化 1s 后执行，之后每隔 3s 上报一次当前机器的信息到 Zookeeper。
private void updateNewData(CuratorFramework curator, String path) {
try {
if (System.currentTimeMillis() < lastUpdateTime) {
return;
}
curator.setData().forPath(path, buildData().getBytes());
lastUpdateTime = System.currentTimeMillis();
} catch (Exception e) {
LOGGER.info("update init data error path is {} error is {}", path, e);
}
}
我们来看下 buildData 的源码，就是看下需要上报什么信息。
/**
* 构建需要上传的数据
*
* @return
*/
private String buildData() throws JsonProcessingException {
Endpoint endpoint = new Endpoint(ip, port, System.currentTimeMillis());
ObjectMapper mapper = new ObjectMapper();
String json = mapper.writeValueAsString(endpoint);
return json;
}
可以看到，就是上报的就是 ip 和 port 和当前时间，这个时间就是 5.2 小节看到的，当 Leaf 启动的时候使用当前时间和 Zookeeper 中最后一次上报的时间做对比，如果发现时钟回退了，就直接返回，因为这种情况下生成的 ID 有可能是重复的。
5.4 updateLocalWorkerID 更新本地文件
这个方法就是在本地节点文件系统上缓存一个 workerID 值，zk 失效，机器重启时保证能够正常启动。
/**
* 在节点文件系统上缓存一个 workid 值,zk 失效,机器重启时保证能够正常启动
*
* @param workerID
*/
private void updateLocalWorkerID(int workerID) {
File leafConfFile = new File(PROP_PATH.replace("{port}", port));
boolean exists = leafConfFile.exists();
LOGGER.info("file exists status is {}", exists);
if (exists) {
try {
FileUtils.writeStringToFile(leafConfFile, "workerID=" + workerID, false);
LOGGER.info("update file cache workerID is {}", workerID);
} catch (IOException e) {
LOGGER.error("update file cache error ", e);
}
} else {
// 不存在文件,父目录页肯定不存在
try {
boolean mkdirs = leafConfFile.getParentFile().mkdirs();
LOGGER.info("init local file cache create parent dis status is {}, worker id is {}", mkdirs, workerID);
if (mkdirs) {
if (leafConfFile.createNewFile()) {
FileUtils.writeStringToFile(leafConfFile, "workerID=" + workerID, false);
LOGGER.info("local file cache workerID is {}", workerID);
}
} else {
LOGGER.warn("create parent dir error===");
}
} catch (IOException e) {
LOGGER.warn("craete workerID conf file error", e);
}
}
}
本地缓存的地址是：${java.io.tmpdir}/${leaf.name}//leafconf/{port}/workerID.properties。
5.5 get 获取 ID
来到最后一个方法，通过 get 获取 ID。
@Override
public synchronized Result get(String key) {
// 获取当前时间
long timestamp = timeGen();
// 时间回拨
if (timestamp < lastTimestamp) {
long offset = lastTimestamp - timestamp;
if (offset <= 5) {
// 如果回拨小于 5ms
try {
// 阻塞等待
wait(offset << 1);
// 再次获取当前时间
timestamp = timeGen();
// 如果还是小于, 说明还是不行
if (timestamp < lastTimestamp) {
return new Result(-1, Status.EXCEPTION);
}
} catch (InterruptedException e) {
LOGGER.error("wait interrupted");
return new Result(-2, Status.EXCEPTION);
}
} else {
return new Result(-3, Status.EXCEPTION);
}
}
// 如果当前时间跟上一次时间相等, 说明在 1ms 内获取了多次 id
if (lastTimestamp == timestamp) {
// 这里 & sequenceMask 是为了当 sequence 递增到 4096 的时候重置成 0
sequence = (sequence + 1) & sequenceMask;
if (sequence == 0) {
// seq 为 0 的时候表示是下一毫秒时间开始对 seq 做随机, 做了随机之后要想通过 id 去查询一些信息比如订单数量什么的就行不通了
sequence = RANDOM.nextInt(100);
// 一直等到下一毫秒
timestamp = tilNextMillis(lastTimestamp);
}
} else {
// 如果是新的 ms 开始
sequence = RANDOM.nextInt(100);
}
// 设置本次访问的时间
lastTimestamp = timestamp;
// 构造 id
long id = ((timestamp - twepoch) << timestampLeftShift) | (workerId << workerIdShift) | sequence;
// 返回结果
return new Result(id, Status.SUCCESS);
}
这里就是雪花算法 ID 生成的逻辑，但是要注意，由于时钟回拨会导致 ID 重复，所以生成之前需要判断如果当前时间小于上一次生成的时间，说明发生了时钟回拨，但是如果回拨不超过 5ms，那么就可以阻塞等待，否则直接返回 Status.EXCEPTION。
同时原来的雪花算法是如果 lastTimestamp != timestamp，就说明到了下一 ms，继续从 0 开始继续生成 ID，但是这里是随机了一下 100 以内的数字，避免连续生成 ID 导致一些业务上面的信息泄露。
其他的就跟雪花算法一样的，感兴趣可以看我的上一篇文章：【分布式 ID】生成唯一 ID 的几种方式，里面有雪花算法的介绍。
5.6 差异
这个图是美团官方的流程图，当然这里获取 Leaf_temporary 下面所有节点，然后做平均值这里代码并没有看到有哪里有写，描述是如果是新服务节点，直接创建持久节点 leaf_forever/${self} 并写入自身系统时间，接下来综合对比其余 Leaf 节点的系统时间来判断自身系统时间是否准确，具体做法是取 leaf_temporary 下的所有临时节点(所有运行中的 Leaf-snowflake 节点)的服务IP：Port，然后通过 RPC 请求得到所有节点的系统时间，计算 sum(time)/nodeSize，可能这个是美团内部正在使用的，又或者是这项目很久没有维护了，如果有知道的朋友也可以说下。
总之大家在看这篇文章的时候主要就是理解两种 ID 生成方式的差异以及优化思想。
6. 小结
好了，这篇文章就到这了，差不多也 3w 字，属实是花了不少篇幅去写源码的部分，主要还是里面的思想，包括说雪花算法如果解决时钟回退的问题，又如何避免生成的 ID 总是 +1 递增，还有就是数据库号段模式下如何使用双 buffer 在消耗完当前 Segment 的号段之后能够不阻塞直接使用下一个号段的 ID。