手写一个高性能分布式 ID 生成器

阿东编程之路 2022-09-04

402

在系统开发中，有非常多的场景会用到唯一 ID 来进行标识，比如订单编号，请求ID，用户ID，数据库分库分表后的主键 ID 等。

先来看下市面上主流的分布式 ID 解决方案~

一. 分布式 ID 业界方案

UUID

UUID 的实现业界有很多种，主流的是基于基于当前时间戳和 mac 地址实现的。

优点：

本地生成，无网络消耗，性能较强。

缺点：

存储空间占用大：UUID 比较长，通常会以 36 长度的字符串表示；
信息不安全：主流的基于 MAC 地址生成的 UUID 会暴露 MAC 地址。
无序性：无序不适合作为 MySQL 的主键和索引，因为 MySQL 的索引是通过 B+ 树实现，无序的索引数据插入会造成整体树结构频繁变动，影响性能。

数据库自增主键

通过数据库的自增主键实现的自增 ID。

优点：自增，不会重复，使用简单。
缺点：强依赖数据库，且性能较差。

Redis 原子自增命令 incr

通过 Redis 的 incr 原子命令自增并返回自增值。

优点：使用简单，性能较强，自增且不会重复。
缺点：强依赖 Redis，如果 Redis 出现故障或发生内存淘汰可能会出现问题。

数据库号段

数据库号段是市面上非常主流的实现分布式ID的方案之一，针对数据自增实现方式采用了「合并请求」优化思想，实现逻辑也比较简单，就是一次向数据库拉取多个 ID，存储在本地，给本地服务提供 ID 发放，当本地用完了，再次向数据库获取一批新 ID，循环往复。

优点：性能强，趋势自增，不会重复。
缺点：还是需要依赖数据库。

雪花算法

雪花算法也是主流的分布式ID实现之一，雪花算法使用 64 位的 long 类型存储 id，最高位存储 0 代表正数，41 位存储时间戳，10 位存储机器码，12 位存储序列号；机器号和序列号长度可以根据业务场景自由定义。

优点：性能强，趋势自增，不会重复。
缺点：强依赖时间戳，如果发生时间回拨或闰秒会出现重复。

美团 Leaf

美团开源了一套对「数据库号段」和「雪花算法」优化后的实现（github: https://github.com/Meituan-Dianping/Leaf）。

数据库号段的优化

普通的数据库号段方式在本地ID用完重新去取时会阻塞等待 IO，针对这点，Leaf 采用了本地双 buffer 的优化，大概逻辑就是，先取出一批 ID 存到第一个 buffer 中，等到 ID 用到 10% 时会去启一个线程异步拉一批新 ID 存储到第二个 buffer 中，循环往复；这种方式有两个有优点：一是能做到取新号段 ID 无阻塞；二是数据库挂了还可以提供服务一段时间。

雪花算法的优化

普通的雪花算法是强依赖时间戳的，如果发布重启时发生时间回拨较大就可能会出现 ID 重复的问题，Leaf 使用 zookeeper 来做注册中心，保证机器号注册唯一，并且每三秒会上报一次系统时间到 zk 上；启动时会去检查上次上报的时间，如果「当前时间」小于上次上报时间超过阈值就启动失败并告警。

二. 自己实现一个高性能分布式 ID 生成器

看过上面一些主流的方案的优化和分析，一个优秀的高性能分布式 ID 有以下要求：

保证全局唯一不重复；

全局趋势递增，保证数据库索引的性能；

尽量减少外部依赖或是外部依赖的时间；

高可用，要有熔断降级的策略来保证出现异常也能对外提供服务；

高性能，ID 生成尽量基于内存；

高并发，在高并发场景下不会有线程安全问题。

综合下来看只有「数据库号段」和「雪花算法」的方式较好，「数据库号段」的方式还是会长时间依赖数据库，最终决定使用「雪花算法」的方式实现。

刚刚上面提到了雪花算法是由 64 位 long 型存储（最高位存储 0 代表正数，41 位存储时间戳，10 位存储机器码，12 位存储序列号），所以保证机器号唯一即可实现分布式下的唯一 ID。业界主流的做法是注册到 zookeeper 上返回顺序节点号当作机器号，这种做法没什么问题，但是时间回拨的解决比较麻烦，每隔三秒需要上报一次系统时间，启动时判断下，如果「当前时间」小于「上次上报时间」超过阈值就只是启动失败并告警。

解决时间回拨造成的 ID 重复有别的思路吗？

我们先看下发生时间回拨的原因：

1. ntp 服务进行时间校准；

2. 发生闰秒，闰秒是受潮汐等自然现象影响造成的地球自转速度变化，为了拥抱这个变化，会向 ntp 服务发送时间回拨的校准。

想要防止这个问题，可以从两个角度考虑：「发生前」和「发生后」：

发生前：有种解决方案是关闭操作系统的ntp服务（时间同步服务），但这种方式不太可控，如果一直不更新时间，本地时间跟正确时间相差就越来越大。

发生后：换种思路，只要最终 ID 不重复就算重启时发生时间回拨也没关系，所以如果发生了时间回拨，我们可以选择重新注册一个新机器号，这个机器号一定得避开上次注册的，这样就可以防止 ID 重复。

机器号的分配和注册

想要 10 位机器号的分配不重复，我们可以选择 Redis 的 setNx 命令来实现服务间的通信，setNx 的逻辑就是：不存在才 set 并返回结果，否则返回 false，这个命令也是 Redis 分布式锁的实现基础。

启动雪花算法服务的时候我们可以随机生成一个机器号（默认0～1023的随机数）注册到 Redis 上，并且这个机器号的生命周期肯定需要大于服务实例的生命周期，为了防止系统异常关闭导致该机器号永远不过期，需要设置过期时间并定时续期，比如每隔 1 分钟续期一次，这样就可以既保证不发生“死锁”又不会有别的实例注册相同的机器号。

续期多久呢？

这个续期时间就是我们解决重启时发生时间回拨的精髓！我们可以根据系统重启的时间来决定续期时间，比如系统启动需要一分钟，我们可以把续期时间设置为 3 分钟，这样重启后，上一个机器号还留在 Redis 上，该服务就会注册新的机器号了。

如果注册机器号失败怎么办？

机器号注册失败是有两种情况，第一种是随机生成的机器号被其他实例注册过，第二种是发生网络异常。我们可以设置两种重试的策略，随机生成策略和顺序生成策略：

随机生成策略：随机生成一个机器号（默认0～1023的随机数）注册到Redis上，如果返回false或异常，再次生成随机数去注册，循环往复，这个循环次数也需要限制，否则可能会造成死循环。

顺序生成策略：从 0 开始向Redis注册机器号，失败进行递增，直到递增至1024。

有可能上述两种策略都失败吗？

如果集群数量小于等于 1024 且 Redis 没挂是不会都失败的，如果两个都失败，也会有个降级的方案：随机生成机器号。

机器号的分配和注册整体逻辑如下：

机器号的分配和注册代码如下：

package com.adong.fingermark.core;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;


/**
 * @author ADong
 * @Description WorkerId 管理，{@link SnowFlakeId#workerId}
 * @Date 2022-08-15 11:45 AM
 */
public class WorkerIdManager {


    private static final Logger log = LoggerFactory.getLogger(WorkerIdManager.class);
    private final RedisTemplate redisTemplate;
    /** 区分业务 **/
    private final String appKey;
    /** workerId 阈值 **/
    private final long maxWorkerId;
    /** 是否开启顺序策略 **/
    private final boolean openSequenceSetWorkerId;
    /** 续期任务线程池 **/
    private final ScheduledExecutorService service;
    /** 续期时间, 单位s **/
    private final long renewalTime;
    /** 续期任务间隔, 单位s **/
    private final long renewalIntervalTime;
    private final Thread shutdownHook;
    /** 机器号 **/
    private volatile Long workerId;


    public WorkerIdManager(RedisTemplate redisTemplate,
                           boolean openSequenceSetWorkerId,
                           String appKey,
                           long maxWorkerId,
                           long renewalTime,
                           long renewalIntervalTime) {
        this.redisTemplate = redisTemplate;
        this.openSequenceSetWorkerId = openSequenceSetWorkerId;
        this.appKey = appKey;
        this.maxWorkerId = maxWorkerId;
        this.renewalTime = renewalTime;
        this.renewalIntervalTime = renewalIntervalTime;
        this.service = Executors.newSingleThreadScheduledExecutor();
        // 钩子函数优雅关闭线程池
        this.shutdownHook = new Thread(() -> destroy());
        Runtime.getRuntime().addShutdownHook(shutdownHook);
        // 续期任务
        renewalWorkerId();
    }


    /**
     * 获取 workerId
     * 先采用随机策略，失败再采用顺序策略，上游给出降级方案
     * @return
     */
    public synchronized Long registerAndGetWorkerId() {
        workerId = null;
        randomSetWorkerId();
        if (openSequenceSetWorkerId && workerId == null) {
            sequenceSetWorkerId();
        }
        return workerId;
    }


    /**
     * 随机策略
     * 取 0～1023 随机数
     * 终止条件：workerId 分配完成 或 或自旋次数大于 10
     */
    private void randomSetWorkerId() {
        int times = 0;
        while (workerId == null && times ++ < 10) {
            long random = ThreadLocalRandom.current().nextLong(0, maxWorkerId);
            workerId = register(random);
        }
    }


    /**
     * 顺序策略
     * 从0开始分配，失败后自增1
     * 终止条件：workerId 分配完成 或 workerId 分配完
     */
    private void sequenceSetWorkerId() {
        long preWorkId = 0;
        while (workerId == null) {
            workerId = register(preWorkId);
            // 如果机器号分配10位，workerId 分配范围 [0,1024)
            if (++ preWorkId >= maxWorkerId) {
                log.error("无可用 workerId！！！");
                return;
            }
        }
    }


    /**
     * 到 Redis 中注册 workId
     * @return 如果失败返回 null，成功直接将 workerId 返回
     */
    private Long register(long workerId) {
        String key = buildSnowFlakeWorkerIdKey(appKey, workerId);
        try {
            if (redisTemplate.opsForValue().setIfAbsent(key, "1", renewalTime, TimeUnit.SECONDS)) {
                log.info("WorkerId={} register success", workerId);
                return workerId;
            }
        } catch (Exception e) {
            log.error("WorkerId register error", e);
        }
        return null;
    }


    /**
     * 续期 workerId
     */
    private void renewalWorkerId()  {
        service.scheduleWithFixedDelay(() -> {
            if (workerId != null) {
                String key = buildSnowFlakeWorkerIdKey(appKey, workerId);
                // 续期时再次set防止被Redis内存淘汰机制清理
                if (!redisTemplate.opsForValue().setIfAbsent(key, "1", renewalTime, TimeUnit.SECONDS)) {
                    redisTemplate.expire(key, renewalTime, TimeUnit.SECONDS);
                }
                log.info("workerId={} 续期成功", workerId);
            }
        }, 3L, renewalIntervalTime, TimeUnit.SECONDS);
    }


    private String buildSnowFlakeWorkerIdKey(String appKey, Long workerId) {
        return String.format("snow_flake_worker_%s_%s", appKey, workerId);
    }


    /**
     * 删除机器号
     * 一般发生时间回拨且集群规模较大的情况下，机器号进行切换后考虑删除上一个机器号
     * @param workerId
     */
    public void delWorkerId(Long workerId) {
        String key = buildSnowFlakeWorkerIdKey(appKey, workerId);
        redisTemplate.delete(key);
    }


    /**
     * 关闭线程池
     */
    public void destroy() {
        service.shutdown();
        log.info("续期线程池关闭");
        // 这里不删除上报机器号防止重启时时间回退造成唯一id发放重复
        // 但是需要保证重启时间小于 renewalTime - renewalIntervalTime
        // 如果后期项目重启时间过长可以适当调大有效期
    }


    public String getAppKey() {
        return appKey;
    }


    public Long getWorkerId() {
        return workerId;
    }
}

刚刚解决的时间回拨是发生在重启时的，那如果发生在服务正常运行时呢？

我们在本地存储上一次使用的时间戳，每次会拿最新生成的「时间戳」和「上次时间戳」进行判断：

如果「上次时间戳」-「最新时间戳」大于 0 且小于等于 5 ms：Object.wait(两倍offset)，等待一会再获取 ID；
如果「上次时间戳」-「最新时间戳」大于 5 ms：重新注册新机器号，再生成ID，即可保证ID不重复（熔断降级）。

雪花算法的代码实现如下：

package com.adong.fingermark.core;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Objects;
import java.util.concurrent.ThreadLocalRandom;


/**
 * @author ADong
 * @Description 参考雪花算法实现分布式id
 * 1位正负位 + 41位时间戳 + 10位机器号 + 12位随机序列
 * 通过 Redis {@link WorkerIdManager#renewalWorkerId()}
 * 续期逻辑保证重启时不会注册上一个机器号来防止重启时的时间回拨问题
 * @Date 2022-08-15 4:24 PM
 */
public class SnowFlakeId implements IdGen {


    private static final Logger log = LoggerFactory.getLogger(SnowFlakeId.class);
    // 记录上次时间戳 41位
    private long lastTime = timeGen();
    // 机房机器ID 10位
    private long workerId = 0;
    private long workerIdShift = 10;
    // 随机数 12位
    private long random = 0;
    private long randomShift = 12;
    // 随机数的最大值
    private long maxRandom = (long) Math.pow(2, randomShift);
    private WorkerIdManager workerIdManager;


    public SnowFlakeId() {}


    public SnowFlakeId(long workerId, long workerIdShift, WorkerIdManager workerIdManager){
        if (workerIdShift < 0 || workerIdShift  > 22) {
            throw new IllegalArgumentException("workerIdShift set error!");
        }
        this.workerId = workerId;
        this.workerIdShift = workerIdShift;
        this.randomShift = 22 - workerIdShift;
        this.maxRandom = (long) Math.pow(2, randomShift);
        this.workerIdManager = workerIdManager;
        log.info("SnowFlakeId init success, workerId={}", workerId);
    }


    // 通过位运算拼接获取 ID
    private long getId() {
        return lastTime << (workerIdShift + randomShift) |
                workerId << randomShift |
                random;
    }
      
    // 降级方案获取 id
    private long getBadId(long badRandom) {
        return lastTime << (workerIdShift + randomShift) |
                badRandom;
    }
    
    // 生成新 ID
    @Override
    public synchronized long nextId() {
        long now = timeGen();
        //如果当前时间大于上次时间，直接返回
        if (now > lastTime) {
            lastTime = now;
            random = getRandom(100);
            return getId();
        }
        // 如果当前时间等于上次时间且random小于最大值随机序列
        if (now == lastTime && ++ random < maxRandom) {
            return getId();
        }
        // 判断如果回拨时间小于5ms或大于等于最大序列值就进行等待，否则进行降级方案
        long offset = lastTime - now == 0 ? 1 : lastTime - now;
        if (offset <= 5) {
            try {
                // 等待两倍offset
                wait(offset << 1);
            } catch (InterruptedException e) {
                log.error("nextId wait interrupted");
                return getBadId(getBadMaxRandom());
            }
        } else {
            // 发生时间回拨切换机器号，失败兜底
            if (!changeWorkerId()) {
                return getBadId(getBadMaxRandom());
            }
        }
        now = timeGen();
        // 再次判断，如果时间还不符合进行降级方案
        if (now < lastTime) {
            return getBadId(getBadMaxRandom());
        } else {
            now = tilNextMillis(lastTime);
        }
        lastTime = now;
        random = getRandom(100);
        return getId();
    }


    private long tilNextMillis(long lastTime) {
        long now = timeGen();
        while (now <= lastTime) {
            now = timeGen();
        }
        return now;
    }


    /**
     * 如果发生时间回拨，重新注册机器号
     * 如果集群规模较大，机器号进行切换后考虑删除上一个机器号 {@link WorkerIdManager#delWorkerId(Long)}
     * @return 是否切换成功
     */
    private boolean changeWorkerId() {
        if (!Objects.isNull(workerIdManager)) {
            Long changeWorkerId = workerIdManager.registerAndGetWorkerId();
            if (changeWorkerId != null) {
                workerId = changeWorkerId;
                lastTime = timeGen();
                // TODO 考虑是否清空上个机器号
                return true;
            }
        }
        return false;
    }


    // 降级方案获取最大随机序列位数
    private long getBadMaxRandom() {
        return getRandom((long) Math.pow(2, randomShift + workerIdShift));
    }


    // 根据最大限制获取随机序列
    private long getRandom(long bound) {
        return ThreadLocalRandom.current().nextLong(bound);
    }


    // 生成当前时间
    private long timeGen() {
        return System.currentTimeMillis();
    }


    @Override
    public void destory() {
        if (!Objects.isNull(workerIdManager)) {
            workerIdManager.destroy();
        }
    }
}

时间回拨的问题我们也解决了，测试下性能吧！

三. 性能对比

测试 UUID 生成一百万个 ID 的耗时：

@Test
public void testSerialUUID() {


    List<String> list = new ArrayList<>();
    Stopwatch stopwatch = Stopwatch.createStarted();
    for (int i = 0; i < 1000000; i++) {
        try {
            list.add(UUID.randomUUID().toString());
        } catch (Exception e) {
            log.error("get id error", e);
        }
    }
    log.info("set size={}, 耗时={}ms", new HashSet<>(list).size(), stopwatch.elapsed(TimeUnit.MILLISECONDS));
}

测试结果：

set size=1000000, 耗时=3171ms

测试阿东改造的雪花算法生成一百万个 ID 的耗时：

@Test
public void testSerialSnowFlakeId() {


    List<Long> list = new ArrayList<>();
    Stopwatch stopwatch = Stopwatch.createStarted();
    for (int i = 0; i < 1000000; i++) {
        try {
            list.add(idGenManager.getId());
        } catch (Exception e) {
            log.error("get id error", e);
        }
    }
    log.info("set size={}, 耗时={}ms", new HashSet<>(list).size(), stopwatch.elapsed(TimeUnit.MILLISECONDS));
}