雪花算法(snowflake)容器化部署支持动态增加节点

Posted 2022-01-04 carl-zhao

先简单的介绍一下雪花算法，雪花算法生成的Id由：1bit 不用 + 41bit时间戳+10bit工作机器id+12bit序列号，如下图：

• 不用：1bit，因为最高位是符号位，0表示正，1表示负，所以这里固定为0
• 时间戳：41bit，服务上线的时间毫秒级的时间戳（为当前时间-服务第一次上线时间），这里为（2^41-1）/1000/60/60/24/365 = 49.7年
• 工作机器id：10bit，表示工作机器id，用于处理分布式部署id不重复问题，可支持2^10 = 1024个节点
• 序列号：12bit，用于离散同一机器同一毫秒级别生成多条Id时，可允许同一毫秒生成2^12 = 4096个Id，则一秒就可生成4096*1000 = 400w个Id

说明：上面总体是64位，具体位数可自行配置，如想运行更久，需要增加时间戳位数；如想支持更多节点，可增加工作机器id位数；如想支持更高并发，增加序列号位数

公司使用的 k8s 容器化部署服务应用，所以需要支持动态增加节点，并且每次部署的机器不一定一样时，就会有问题。参考了 雪花算法snowflake生成Id重复问题 其中的思想：

• 在redis中存储一个当前workerId的最大值
• 每次生成workerId时，从redis中获取到当前workerId最大值，并+1作为当前workerId，并存入redis
• 如果workerId为1023，自增为1024，则重置0，作为当前workerId，并存入redis

然后优化成以下逻辑：

• 定义一个 redis 作为缓存 key，然后服务每次初始化的时候都 incr 这个 key。
• 上面得到的 incr 的结果然后与 1024 取模。取模可以优化为：result & 0x000003FF

所以最后的代码为下面：

首先我们先定义雪花算法生成分布式 ID 类：

SnowflakeIdWorker.java

public class SnowflakeIdWorker 
    /** 开始时间截 (建议用服务第一次上线的时间，到毫秒级的时间戳) */
    private final long twepoch = 687888001020L;

    /** 机器id所占的位数 */
    private final long workerIdBits = 10L;

    /** 支持的最大机器id，结果是1023 (这个移位算法可以很快的计算出几位二进制数所能表示的最大十进制数) */
    private final long maxWorkerId = -1L ^ (-1L << workerIdBits);

    /** 序列在id中占的位数 */
    private final long sequenceBits = 12L;

    /** 机器ID向左移12位 */
    private final long workerIdShift = sequenceBits;

    /** 时间截向左移22位(10+12) */
    private final long timestampLeftShift = sequenceBits + workerIdBits;

    /** 生成序列的掩码，这里为4095 (0b111111111111=0xfff=4095)
     * <<为左移，每左移动1位，则扩大1倍
     * */
    private final long sequenceMask = -1L ^ (-1L << sequenceBits);

    /** 工作机器ID(0~1024) */
    private long workerId;

    /** 毫秒内序列(0~4095) */
    private long sequence = 0L;

    /** 上次生成ID的时间截 */
    private long lastTimestamp = -1L;

    //==============================Constructors=====================================
    /**
     * 构造函数
     * @param workerId 工作ID (0~1023)
     */
    public SnowflakeIdWorker(long workerId) 
        if (workerId > maxWorkerId || workerId < 0) 
            throw new IllegalArgumentException(String.format("workerId can't be greater than %d or less than 0", maxWorkerId));
        
        this.workerId = workerId;
    

    // ==============================Methods==========================================
    /**
     * 获得下一个ID (该方法是线程安全的)
     * @return SnowflakeId
     */
    public synchronized long nextId() 
        long timestamp = timeGen();
        //如果当前时间小于上一次ID生成的时间戳，说明系统时钟回退过这个时候应当抛出异常
        if (timestamp < lastTimestamp) 
            throw new RuntimeException(
                    String.format("Clock moved backwards.  Refusing to generate id for %d milliseconds", lastTimestamp - timestamp));
        

        //如果是同一时间生成的，则进行毫秒内序列
        if (lastTimestamp == timestamp) 
            //如果毫秒相同，则从0递增生成序列号
            sequence = (sequence + 1) & sequenceMask;
            //毫秒内序列溢出
            if (sequence == 0) 
                //阻塞到下一个毫秒,获得新的时间戳
                timestamp = tilNextMillis(lastTimestamp);
            
        
        //时间戳改变，毫秒内序列重置
        else 
            sequence = 0L;
        

        //上次生成ID的时间截
        lastTimestamp = timestamp;

        //移位并通过或运算拼到一起组成64位的ID
        return ((timestamp - twepoch) << timestampLeftShift) //
                | (workerId << workerIdShift) //
                | sequence;
    

    /**
     * 阻塞到下一个毫秒，直到获得新的时间戳
     * @param lastTimestamp 上次生成ID的时间截
     * @return 当前时间戳
     */
    protected long tilNextMillis(long lastTimestamp) 
        long timestamp = timeGen();
        while (timestamp <= lastTimestamp) 
            timestamp = timeGen();
        
        return timestamp;
    

    /**
     * 返回以毫秒为单位的当前时间，从1970-01-01 08:00:00算起
     * @return 当前时间(毫秒)
     */
    protected long timeGen() 
        return System.currentTimeMillis();
    

    public static void main(String[] args) 
        SnowflakeIdWorker snowflakeIdWorker = new SnowflakeIdWorker(1);
        Set<Long> params = new HashSet<>();
        for (int i = 0; i < 3000_0000; i++) 
            params.add(snowflakeIdWorker.nextId());
        
        System.out.println(params.size());
    

接着定义一个 ID 生成的接口以及实现类。

public interface IdManager 

    String getId();



 下面是实现类

@Slf4j
@Service("idManager")
public class IdManagerImpl implements IdManager 

    @Resource(name = "stringRedisTemplate")
    private StringRedisTemplate stringRedisTemplate;

    private SnowflakeIdWorker snowflakeIdWorker;

    @PostConstruct
    public void init() 
        String cacheKey = KeyUtils.getKey("order", "snowflake", "workerId", "incr");

        Long increment = stringRedisTemplate.opsForValue().increment(cacheKey);

        long workerId = increment & 0x000003FF;
        log.info("IdManagerImpl.init snowflake worker id is ", workerId);

        snowflakeIdWorker = new SnowflakeIdWorker(workerId);
    

    @Override
    public String getId() 
        long nextId = snowflakeIdWorker.nextId();

        return Long.toString(nextId);
    

在服务每次上线的时候就会把之前的 incr 值加 1。然后与 1024 取模，最后 workerId 就会一直在 [0 ~ 1023] 范围内进行动态取值。