深度解析leaf分布式id生成服务源码（snowflake模式）

妖狐艹你老母 2022-10-13 12:36 181阅读 0赞

> 原创不易，转载请注明出处

### 文章目录 ###

*   *   *  前言
         *  1.snowflake雪花id算法原理
         *  2.单纯使用snowflake存在的问题
         *   *  2.1 workerId问题
             *  2.2 时钟回拨问题
         *  3.leaf对于snowflake存在问题的解决方案
         *   *  3.1基于zk的顺序节点生成workerId
             *  3.2记录上一个时间戳信息
         *  4.snowflake模式源码分析
         *   *  4.1基于zk的workerId实现
             *  4.2 记录上一个时间戳信息

--------------------

### 前言 ###

在《深度解析leaf分布式id生成服务源码（号段模式）》一文中我们深度剖析了leaf的号段模式底层实现原理与源码，本文接着剖析下leaf的另一个模式snowflake模式的实现原理与源码。

### 1.snowflake雪花id算法原理 ###

关于雪花算法网上一搜一大把，它是由推特公司开源的分布式id生成算法。这个分布式id生成不像号段模式那样依赖mysql这种外部资源，它是一套算法，可以放到任何系统中（一般不这么干，与业务系统耦合了）。  
我们接下来说下这个雪花算法的组成部分吧。  
它生成的是一个64位long类型的数字。  
64位分为4部分  
![在这里插入图片描述][20210706233312758.png]  
1-12位代表的是 序号 是12位，最多4096个数  
13-22位是机器id 是10位，最多1024  
23-63 位表示时间戳 是41位，最多也就能表示69年的样子。  
64位是符号位，这个不用管。

这个样子雪花id生成规则就是  
时间戳+ 机器id+序号  
用这套算法的话，一台机器每毫秒最多允许生成4096个不重复的id，前提是你workerId也就是机器id不能重复。

### 2.单纯使用snowflake存在的问题 ###

#### 2.1 workerId问题 ####

这里的workerId要求是不能重复的，如果重复的话，就有很大机率生成重复id，所以说保证这个workerId不重复是个问题，当然也就人说，我认为维护这个workerId，服务数量少还好说，服务数量多了你就整不过来了。

#### 2.2 时钟回拨问题 ####

可以看到雪花id这套算法是强依赖操作系统的时间戳的，我们服务器也会进行时间同步，这个时间同步容易造成时钟回拨的现象。  
也就是会出现这种情况，比如说现在11点12分10秒的300毫秒，时钟同步一下可能会变成11点12分10秒的100毫秒，也就是服务器时间发生回退。这样子生成的分布式id就可能发生重复问题。

### 3.leaf对于snowflake存在问题的解决方案 ###

#### 3.1基于zk的顺序节点生成workerId ####

leaf是workerId是依赖zk的顺序节点生成的，你会发现你配置leaf的时候会配置一个port的参数，它正是通过你本机ip+你配置的那个端口在生成一个永久顺序节点。逻辑大体是这个样子的，先会判断ip+port对应的那个永久顺序节点是否存在，如果不存在的话，就会创建一个，根据zk顺序节点的特性，会在你节点后面生成一串序号，比如说 127.0.0.1:1000-000000001这个样子，这里0000001就是它的序号，它其实就是取这个序号作为workerId，如果存在的话，就获取所有的序号，找出你这个ip:port对应的那个序号，它同时也会将workerId持久化到磁盘上面。

#### 3.2记录上一个时间戳信息 ####

通过记录上次时间戳与当前时间戳做比较，如果说当前时间戳小于上次时间戳，说明发生了时钟回拨，它这里主要是看看回拨的一个程度，如果说是小于5ms，就等wait一下，如果是大于这个5ms的话，就对客户端返回异常了，其实这里就需要客户端配合去其他leaf服务上面获取，或者是重试获取等等。

### 4.snowflake模式源码分析 ###

#### 4.1基于zk的workerId实现 ####

项目启动的时候会实例化SnowflakeIDGenImpl，在SnowflakeIDGenImpl构造中会创建一个SnowflakeZookeeperHolder对象，然后调用其init方法进行初始化。我们看下SnowflakeZookeeperHolder\#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/leaf.name/forever/ip:port-000000000,并上传数据
                    //创建永久顺序节点
                    zk_AddressNode = createNode(curator);
                    //worker id 默认是0
                    //将workerId 搞到本地文件中
                    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]));
                    }
                    //获取listenAddress 的workerId
                    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);
                        // 更新本地文件中的workerId信息
                        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("-");
                        workerID = Integer.parseInt(nodeKey[1]);
                        doService(curator);
                        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 { 
                    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;

代码非常长，我们就不行行看了，这里解释下代码逻辑，开始就是检查/snowflake/leaf.name/forever 这个节点时候存在。如果不存在的话，说明之前没有用过。这个时候它会创建一个这个服务的永久顺序节点，比如说当前这个leaf服务的ip是127.0.0.1 配置的port是10005  
这个时候它会创建这么一个节点  
/snowflake/leaf.name/forever/127.0.0.1:10005-0000001  
0000001这个序号就是这个服务的workerId，同时将workerId写到磁盘文件中，启动一个延时任务，每3s上报一次当前系统时间。  
如果说/snowflake/leaf.name/forever 这个永久节点已经存在了，说明这个时候有其他leaf服务生成过workerId，这个时候就会将/snowflake/leaf.name/forever这个节点的所有子节点拿到，进行遍历比对，找出127.0.0.1:10005对应的那个workerId，如果没有的话，也会创建这个顺序节点，持久化到本地磁盘workerId，启动延时任务。  
如果找到了的话，会判断一下当前系统时间是否是大于zk这个节点之前上传的那个时间戳，不大于的话说明时钟回拨严重不让启动。  
大于的话，就会启动延时任务，3s上传到zk 当前系统时间戳。

#### 4.2 记录上一个时间戳信息 ####

接下来看下它基于雪花算法+时钟回拨优化的代码实现。  
直接找到SnowflakeIDGenImpl 的get方法

@Override
        public synchronized Result get(String key) { 
    
            // 获取系统时间戳
            long timestamp = timeGen();
            //如果小于上次的话，说明时钟回拨
            if (timestamp < lastTimestamp) { 
    
                //计算差值
                long offset = lastTimestamp - timestamp;
                //差值小于等于5毫秒
                if (offset <= 5) { 
                    try { 
                        //等待 2倍差值时间
                        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);
                }
            }
    
            // 上次时间与这次的一样
            if (lastTimestamp == timestamp) { 
    
                // 00001111 11111111
                // 00010000 00000000
    
                sequence = (sequence + 1) & sequenceMask;
                // seq 消耗完的时候。切换到下一个毫秒
                if (sequence == 0) { 
                    //seq 为0的时候表示是下一毫秒时间开始对seq做随机
                    sequence = RANDOM.nextInt(100);
                    timestamp = tilNextMillis(lastTimestamp);
                }
            } else { 
                //如果是新的ms开始
                sequence = RANDOM.nextInt(100);
            }
            lastTimestamp = timestamp;
            long id = ((timestamp - twepoch) << timestampLeftShift) | (workerId << workerIdShift) | sequence;
            return new Result(id, Status.SUCCESS);
    
        }

代码也是很长，先是获取当前系统时间戳，与lastTimestamp 也就是上次时间戳做对比，如果小于的话，说明是时钟回拨，就会计算差值，如果是小于5ms的话，直接wait双倍，之后继续获取当前时间戳，如果是还是小于的话，就返回异常了。  
之后就是当前时间戳与lastTimestamp 相等的时候，计算下seq ，如果是超了4095个，这个时候，也会等，等到下一毫秒，然后重置下seq，后面的就是雪花算法那一套了，这里就不赘述。

[20210706233312758.png]: /images/20221005/3aab8326fcf342aa8a265348baab253d.png