分布式唯一ID生成（二）: leaf

本系列

• 漫谈分布式唯一ID
• 分布式唯一ID生成（二）：leaf（本文）
• 分布式唯一ID生成（三）：uid-generator
• 分布式唯一ID生成（四）：tinyid

前言

本文将介绍leaf号段模式和雪花算法模式的设计原理，并走读源码

源码地址：https://github.com/Meituan-Dianping/Leaf

leaf提供了 leaf server，业务只管调leaf server的接口获取ID，leaf serve内部根据号段或雪花算法生成ID，而不是业务服务自己去请求数据库生成id，或自己根据雪花算法生成id

号段模式

在使用db自增主键的基础上，从每次获取ID都要读写一次db，改成一次获取一批ID

各个业务的记录通过 biz_tag 区分，每个业务的ID上次分配到哪了，在一张表中用一条记录表示

表结构如下：

CREATE TABLE `leaf_alloc` (`biz_tag` varchar(128)  NOT NULL DEFAULT '', -- your biz unique name`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;

重要字段为：

• biz_tag：标识业务
• max_id：目前分配到的最大值-1，也是 下一个号段的起始值
• step：每次分配号段的长度

如果step=1000，当这1000个ID消耗完后才会读写一次DB，对DB的压力降为原来的 1/1000

当缓存中没有ID时，需要从db获取号段，在事务中执行如下2条sql：

UPDATE leaf_alloc SET max_id = max_id + step WHERE biz_tag = #{tag}

SELECT biz_tag, max_id, step FROM leaf_alloc WHERE biz_tag = #{tag}

然后加载到本地

N个server执行上述操作，对外提供http接口用于生成id，整体架构如下图所示：

优点：

• Leaf服务可以很方便的线性扩展，例如按照biz_tag分库分表
• ID是趋势递增的64位数字，满足上述数据库存储的类型要求
• 容灾性高：Leaf服务内部有号段缓存，即使DB宕机，短时间内仍能正常对外提供服务
• 易接入：可自定义初始max_id的值，方便业务从原有的ID方式上迁移过来

缺点：

• 业务上不够安全：ID近似于严格递增，会泄露发号数量的信息

• TP999显著比其他TP值大：当号段使用完之后还是会hang在更新数据库的I/O上

  • 以step=1000为例，99.9%的请求分配ID都非常快，0.1%的请求会比较慢（读写一次db平均5ms），如果恰好遇到db抖动，耗时能到几秒

• 单点问题：DB如果宕机会造成整个系统不可用

• 网络IO开销大：client每获取一个id，都要对leaf server发起一次http调用

双buffer优化

针对TP999大的问题，Leaf号段模式做了一些优化：在内存中维护两个号段，在当前号段消费到一定百分比时，就 异步去db加载下一个号段 到内存中
这样当前号段用完后，就能马上切换到下一个号段

biz优化

对于每个请求，都需要校验参数中的biz是否合法。如果每次都去db查下biz在leaf_alloc表是否存在，性能开销大且没必要

leaf在实例启动时，将全量biz都查出来放到本地缓存中，之后每隔60s都会刷新一次，这样校验biz是否合法都用本地map判断，性能极高

缺点是最多延迟1分钟才新增的biz才生效

也就牺牲一点一致性换取超高的性能

动态step

如果每次获取号段的长度step是固定的，但流量不是固定的，如果流量增加 10 倍，每个号段很快就被用完了，仍然有可能导致数据库压力较大

同时也降低了可用性，例如本来能在DB不可用的情况下维持10分钟正常工作，那么如果流量增加10倍就只能维持1分钟正常工作了

因此leaf中每次从db加载号段时，加载多少ID并不是固定的

• 如果qps高，就可以一次多加载点，减少调db的次数
• 如果qps低，可以一次少加载点。否则在缓存中的号段迟迟消耗不完的情况下，会导致更新DB的新号段与前一次下发的号段ID跨度过大

leaf的策略为每次更新buffer时动态维护step，当需要从db加载号段时，计算距离上次从db加载过去了多久：

• 小于15mins：获取的号段长度翻倍
• 15～30mins：获取的号段长度和上次一样
• 大于30mins：获取的号段长度减半

源码走读

初始化：

public boolean init() {// 将所有biz加载到内存updateCacheFromDb();initOK = true;// 后台每1s刷新一次内存中的bizupdateCacheFromDbAtEveryMinute();return initOK;
}

获取ID流程如下：

public Result get(final String key) {if (!initOK) {return new Result(EXCEPTION_ID_IDCACHE_INIT_FALSE, Status.EXCEPTION);}// 校验biz是否合法if (cache.containsKey(key)) {SegmentBuffer buffer = cache.get(key);if (!buffer.isInitOk()) {synchronized (buffer) {if (!buffer.isInitOk()) {try {// 号段未初始化，从db加载号段updateSegmentFromDb(key, buffer.getCurrent());buffer.setInitOk(true);} catch (Exception e) {}}}}// 从号段获取IDreturn getIdFromSegmentBuffer(cache.get(key));}return new Result(EXCEPTION_ID_KEY_NOT_EXISTS, Status.EXCEPTION);
}

getIdFromSegmentBuffer方法：

public Result getIdFromSegmentBuffer(final SegmentBuffer buffer) {while (true) {buffer.rLock().lock();try {final Segment segment = buffer.getCurrent();// 如果当前号段已经用了10%，异步去加载下一个号段if (!buffer.isNextReady() && (segment.getIdle() < 0.9 * segment.getStep()) && buffer.getThreadRunning().compareAndSet(false, true)) {service.execute(new Runnable() {@Overridepublic void run() {// 加载下一个号段});}// 当前号段还没用完，从当前号段获取long value = segment.getValue().getAndIncrement();if (value < segment.getMax()) {return new Result(value, Status.SUCCESS);}} finally {buffer.rLock().unlock();}// 到这说明当前号段用完了waitAndSleep(buffer);buffer.wLock().lock();try {// 再次检查当前号段，因为可能别的线程加载了final Segment segment = buffer.getCurrent();long value = segment.getValue().getAndIncrement();if (value < segment.getMax()) {return new Result(value, Status.SUCCESS);}// 切换到下一个号段，重新执行while循环获取if (buffer.isNextReady()) {buffer.switchPos();buffer.setNextReady(false);} else {// 两个号段都不可用，报错return new Result(EXCEPTION_ID_TWO_SEGMENTS_ARE_NULL, Status.EXCEPTION);}} finally {buffer.wLock().unlock();}}}

最后看下怎么从db加载号段：

public void updateSegmentFromDb(String key, Segment segment) {SegmentBuffer buffer = segment.getBuffer();LeafAlloc leafAlloc;// ...// 动态调整下一次的steplong duration = System.currentTimeMillis() - buffer.getUpdateTimestamp();int nextStep = buffer.getStep();if (duration < SEGMENT_DURATION) {if (nextStep * 2 > MAX_STEP) {} else {nextStep = nextStep * 2;}} else if (duration < SEGMENT_DURATION * 2) {} else {nextStep = nextStep / 2 >= buffer.getMinStep() ? nextStep / 2 : nextStep;}LeafAlloc temp = new LeafAlloc();temp.setKey(key);temp.setStep(nextStep);// 从db获取下一个号段leafAlloc = dao.updateMaxIdByCustomStepAndGetLeafAlloc(temp);buffer.setUpdateTimestamp(System.currentTimeMillis());buffer.setStep(nextStep);buffer.setMinStep(leafAlloc.getStep());// 加载到内存中long value = leafAlloc.getMaxId() - buffer.getStep();segment.getValue().set(value);segment.setMax(leafAlloc.getMaxId());segment.setStep(buffer.getStep());
}

内存中Segment结构主要有以下字段：

• value：下一个要分配的ID
• max：当前号段的最大边界

每次从Segment中分配ID时，返回value的值即可，并把value++

雪花算法

号段模式的ID很接近严格递增，如果在订单场景，可以根据ID猜到一天的订单量。此时就可以用雪花算法模式

leaf在每一位的分配和标准snowflake一致：

• 最高位符号位为0

• 接下来41位：毫秒级时间戳

  • 存储当前时间距离2010年某一天的差值

• 接下来10位：workerId

• 最后12位：每一毫秒内的序列号

每到新的毫秒时，每一毫秒内的序列号不是从0开始，而是从100以内的一个随机数开始

为啥这么设计？试想如果每一秒都从0开始，在qps低的情况下，每一毫秒只产生1个id，那么最末尾永远是0。如果对ID取模分表，就会永远在第0号表，造成数据分布不均

怎么设置workerId

对于workerID的分配，当服务集群数量较小的情况下，完全可以手动配置。如果服务规模较大，动手配置成本太高。于是leaf用zookeeper 自动获取workerId，流程如下：

1. 以自己的 ip:port为key，去zk建立持久顺序节点，以zk生成的 自增序号为workerId

   1. 创建的节点最后两级路径为：/forever/ip:port-序列号

2. 如果zk中已经有自己ip:port的节点，就 复用 其workerId

   1. 怎么判断？拉取/forever下所有节点，每个节点的格式为ipport-序列号，判断每个节点中-前面的ipport是不是等于自己，如果等于取-后面的序列号作为workerId
   2. 只有leaf server会创建zk节点，因此节点数量可控
   3. 为啥可以复用？不会在同一时刻，有相同ip:port的两个实例，因此复用一定不会发生冲突

这种workerId分配策略能保证唯一性吗？能

• 如果 ip:port 不同，在zk中一定是两个不同的序列号，因此不会冲突
• 一个集群中不可能同时存在ip:port相同的两个机器

每个leaf server的ip:port最好手动指定，或者部署在ip不会变化的环境中

高可用：workerId会存到本地文件，这样遇到极端情况：leaf server服务重启，且zk也宕机时，也不影响使用

解决时钟回拨

雪花算法严格依赖时间，如果发生了时钟回拨，就可能导致ID重复，因此需要监测是否发生了时钟回拨并处理，在服务启动和运行时都会检测

在服务启动时检测时间是否回退：

• leaf server运行时，每隔3s会上自己的当前时间到zk节点中
• 启动时，校验当前时间不能小于 zk中最近一次上报的时间

官方文档还提到如果是第一次启动，还会和其他leaf server校准时间。但源码中没找到这部分，应该是不需要做这个校准，已删除

在运行时检测时间是否回退：

• 全局维护了上次获取ID时的时间戳：lastTimestamp

• 如果当前时间 now < lastTimestamp，说明发生了时钟回拨

  • 回拨了超过5ms，返回报错
  • 回拨了5ms内，sleep一会，直到赶上上次时间

如果zk宕机导致定时上报没有成功，同时又发生了时钟回拨，且leaf server宕机。此时leaf server启动时可能产生和之前重复的ID。因此需要做好监控告警，zk的高可用

如果3s内没上报，leaf server宕机了，然后时钟回退了2s，此时根据zk的时间检测不出来发生了时钟回退，也会造成ID重复。解决方法就是等一段时间才重启机器，保证等待的时间比回拨的时间长就行

源码走读

初始化：

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);//每3s上报本机时间给forever节点ScheduledUploadData(curator, zk_AddressNode);return true;} else {Map<String, Integer> nodeMap = Maps.newHashMap();//ip:port->00001Map<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:portzk_AddressNode = PATH_FOREVER + "/" + realNode.get(listenAddress);workerID = workerid;//启动worder时使用会使用// 当前时间不能小于 zk中最近一次上报的时间if (!checkInitTimeStamp(curator, zk_AddressNode)) {throw new CheckLastTimeException("init timestamp check error,forever node timestamp gt this node time");}// 每3s上报时间doService(curator);// 将workerId写到本地updateLocalWorkerID(workerID);} else {//新启动的节点,创建持久节点 ,不用check时间String newNode = createNode(curator);zk_AddressNode = newNode;String[] nodeKey = newNode.split("-");workerID = Integer.parseInt(nodeKey[1]);doService(curator);updateLocalWorkerID(workerID);}}} catch (Exception e) {// zk不可用，从本地文件加载workerIdtry {Properties properties = new Properties();properties.load(new FileInputStream(new File(PROP_PATH.replace("{port}", port + ""))));workerID = Integer.valueOf(properties.getProperty("workerID"));} catch (Exception e1) {return false;}}return true;}

获取ID：

public synchronized Result get(String key) {long timestamp = timeGen();// 发生了时钟回拨if (timestamp < lastTimestamp) {long offset = lastTimestamp - timestamp;// 回拨了5ms内，sleep一会if (offset <= 5) {try {wait(offset << 1);timestamp = timeGen();if (timestamp < lastTimestamp) {return new Result(-1, Status.EXCEPTION);}} catch (InterruptedException e) {return new Result(-2, Status.EXCEPTION);}// 回拨了超过5ms，返回报错} else {return new Result(-3, Status.EXCEPTION);}}// 和上次在同一毫秒if (lastTimestamp == timestamp) {sequence = (sequence + 1) & sequenceMask;if (sequence == 0) {//seq 为0表示是当前ms已经超过4096个ID了// 需要sleep一会，下一毫秒时间开始对seq做随机sequence = RANDOM.nextInt(100);timestamp = tilNextMillis(lastTimestamp);}} else {//如果是新的ms, 对seq做随机sequence = RANDOM.nextInt(100);}lastTimestamp = timestamp;long id = ((timestamp - twepoch) << timestampLeftShift) | (workerId << workerIdShift) | sequence;return new Result(id, Status.SUCCESS);}

总结

leaf提供两种分布式ID生成策略：

• 号段模式：

  • 每次从db获取一批ID，而不是一个ID，减少调DB的频率
  • 用双buffer解决TP999耗时高的问题
  • 在内存判断参数biz是否合法，提高校验性能
  • 使用动态step，解决突发流量造成对db压力仍然大的问题

• 雪花算法模式：

  • 配合ZK做到动态获取workerId，解决海量机器的 workId 维护问题，也能保证正确性：同时不会有两个leaf server拥有相同的workerId
  • 在服务启动时和运行时都校验是否发生了时钟回拨。不过服务启动时的校验有时会失效，最好sleep一段时间再重启，这段时间要大于时钟回拨的时间