美团开源分布式ID生成系统——Leaf源码阅读笔记（Leaf的号段模式）

末蓝、 2023-06-10 15:28 180阅读 0赞

Leaf 最早期需求是各个业务线的订单ID生成需求。在美团早期，有的业务直接通过DB自增的方式生成ID，有的业务通过redis缓存来生成ID，也有的业务直接用UUID这种方式来生成ID。以上的方式各自有各自的问题，因此我们决定实现一套分布式ID生成服务来满足需求。具体Leaf 设计文档见： [ leaf 美团分布式ID生成服务 ][leaf _ID_]。

出于对该中间件的研究笔者建议下载源码阅读：[github源码地址][github]。

Leaf提供两种模式来生产ID，分别是号段模式（数据库递增）和snowflake模式。  
本文针对数据库生产的号段模式的源码来研究。

以MySQL举例，利用给字段设置auto\_increment\_increment和auto\_increment\_offset来保证ID自增，每次业务使用下列SQL读写MySQL得到ID号。

begin;
    REPLACE INTO Tickets64 (stub) VALUES ('a');
    SELECT LAST_INSERT_ID();
    commit;

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Nhbm90_size_16_color_FFFFFF_t_70]

这种方案的优缺点如下：

优点：

>  *  非常简单，利用现有数据库系统的功能实现，成本小，有DBA专业维护。
>  *  ID号单调自增，可以实现一些对ID有特殊要求的业务。  
>     缺点：

强依赖DB，当DB异常时整个系统不可用，属于致命问题。配置主从复制可以尽可能的增加可用性，但是数据一致性在特殊情况下难以保证。主从切换时的不一致可能会导致重复发号。

ID发号性能瓶颈限制在单台MySQL的读写性能。

对于MySQL性能问题，可用如下方案解决：在分布式系统中我们可以多部署几台机器，每台机器设置不同的初始值，且步长和机器数相等。比如有两台机器。设置步长step为2，TicketServer1的初始值为1（1，3，5，7，9，11…）、TicketServer2的初始值为2（2，4，6，8，10…）。这是Flickr团队在2010年撰文介绍的一种主键生成策略（[Ticket Servers: Distributed Unique Primary Keys on the Cheap][Ticket Servers_ Distributed Unique Primary Keys on the Cheap] ）。如下所示，为了实现上述方案分别设置两台机器对应的参数，TicketServer1从1开始发号，TicketServer2从2开始发号，两台机器每次发号之后都递增2。

TicketServer1:
    auto-increment-increment = 2
    auto-increment-offset = 1
    
    TicketServer2:
    auto-increment-increment = 2
    auto-increment-offset = 2

这种架构貌似能够满足性能的需求，但有以下几个缺点：

系统水平扩展比较困难，比如定义好了步长和机器台数之后，如果要添加机器该怎么做？假设现在只有一台机器发号是1,2,3,4,5（步长是1），这个时候需要扩容机器一台。可以这样做：把第二台机器的初始值设置得比第一台超过很多，比如14（假设在扩容时间之内第一台不可能发到14），同时设置步长为2，那么这台机器下发的号码都是14以后的偶数。然后摘掉第一台，把ID值保留为奇数，比如7，然后修改第一台的步长为2。让它符合我们定义的号段标准，对于这个例子来说就是让第一台以后只能产生奇数。扩容方案看起来复杂吗？貌似还好，现在想象一下如果我们线上有100台机器，这个时候要扩容该怎么做？简直是噩梦。所以系统水平扩展方案复杂难以实现。

ID没有了单调递增的特性，只能趋势递增，这个缺点对于一般业务需求不是很重要，可以容忍。

数据库压力还是很大，每次获取ID都得读写一次数据库，只能靠堆机器来提高性能。

对比上述几种方案，每种方案都不完全符合要求。Leaf在方案上做了相应的优化，实现了Leaf-segment方案。

### Leaf-segment数据库方案 ###

第一种Leaf-segment方案，在使用数据库的方案上，做了如下改变：

原方案每次获取ID都得读写一次数据库，造成数据库压力大。改为利用proxy server批量获取，每次获取一个segment(step决定大小)号段的值。用完之后再去数据库获取新的号段，可以大大的减轻数据库的压力。

各个业务不同的发号需求用biz\_tag字段来区分，每个biz-tag的ID获取相互隔离，互不影响。如果以后有性能需求需要对数据库扩容，不需要上述描述的复杂的扩容操作，只需要对biz\_tag分库分表就行。

数据库表设计如下：

CREATE TABLE `leaf_alloc` (
      `biz_tag` varchar(128) NOT NULL DEFAULT '' COMMENT '业务编号，不同对业务编号的id生产互相隔离' ,
      `max_id` bigint(20) NOT NULL DEFAULT '1' COMMENT '当前的最大id',
      `step` int(11) NOT NULL  COMMENT '号段的范围，即一次加载到内存中可以生产的id值的数量。该值不易过小，否则会频繁更新sql',
      `description` varchar(256) DEFAULT NULL,
      `update_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
      PRIMARY KEY (`biz_tag`)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8;

biz\_tag用来区分业务，max\_id表示该biz\_tag目前所被分配的ID号段的最大值，step表示每次分配的号段长度。原来获取ID每次都需要写数据库，现在只需要把step设置得足够大，比如1000。那么只有当1000个号被消耗完了之后才会去重新读写一次数据库。读写数据库的频率从1减小到了1/step，大致架构如下图所示：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Nhbm90_size_16_color_FFFFFF_t_70 1]

test\_tag在第一台Leaf机器上是1～1000的号段，当这个号段用完时，会去加载另一个长度为step=1000的号段，假设另外两台号段都没有更新，这个时候第一台机器新加载的号段就应该是3001~4000。同时数据库对应的biz\_tag这条数据的max\_id会从3000被更新成4000，更新号段的SQL语句如下：

Begin
    UPDATE table SET max_id=max_id+step WHERE biz_tag=xxx
    SELECT tag, max_id, step FROM table WHERE biz_tag=xxx
    Commit

优点：

>  *  Leaf服务可以很方便的线性扩展，性能完全能够支撑大多数业务场景。
>  *  ID号码是趋势递增的8byte的64位数字，满足上述数据库存储的主键要求。
>  *  容灾性高：Leaf服务内部有号段缓存，即使DB宕机，短时间内Leaf仍能正常对外提供服务。
>  *  可以自定义max\_id的大小，非常方便业务从原有的ID方式上迁移过来。

### 双buffer优化 ###

对于第二个缺点，Leaf-segment做了一些优化，简单的说就是：

Leaf 取号段的时机是在号段消耗完的时候进行的，也就意味着号段临界点的ID下发时间取决于下一次从DB取回号段的时间，并且在这期间进来的请求也会因为DB号段没有取回来，导致线程阻塞。如果请求DB的网络和DB的性能稳定，这种情况对系统的影响是不大的，但是假如取DB的时候网络发生抖动，或者DB发生慢查询就会导致整个系统的响应时间变慢。

为此，我们希望DB取号段的过程能够做到无阻塞，不需要在DB取号段的时候阻塞请求线程，即当号段消费到某个点时就异步的把下一个号段加载到内存中。而不需要等到号段用尽的时候才去更新号段。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Nhbm90_size_16_color_FFFFFF_t_70 2]

采用双buffer的方式，Leaf服务内部有两个号段缓存区segment。当前号段已下发10%时，如果下一个号段未更新，则另启一个更新线程去更新下一个号段。当前号段全部下发完后，如果下个号段准备好了则切换到下个号段为当前segment接着下发，循环往复。

> 每个biz-tag都有消费速度监控，通常推荐segment长度设置为服务高峰期发号QPS的600倍（10分钟），这样即使DB宕机，Leaf仍能持续发号10-20分钟不受影响。

> 每次请求来临时都会判断下个号段的状态，从而更新此号段，所以偶尔的网络抖动不会影响下个号段的更新。

### 源码解读 ###

**返回生成的id包装类：**

public class Result {
        private long id;
        private Status status;
    
        public Result() {
    
        }
        public Result(long id, Status status) {
            this.id = id;
            this.status = status;
        }
    
        public long getId() {
            return id;
        }
    
        public void setId(long id) {
            this.id = id;
        }
    
        public Status getStatus() {
            return status;
        }
    
        public void setStatus(Status status) {
            this.status = status;
        }
    
        @Override
        public String toString() {
            final StringBuilder sb = new StringBuilder("Result{");
            sb.append("id=").append(id);
            sb.append(", status=").append(status);
            sb.append('}');
            return sb.toString();
        }
    }

两种生成id的模式的通用接口

public interface IDGen {
        Result get(String key);
        boolean init();
    }

buffer类解读：

public class Segment {
        /**
         * 当前ID，下一个id执行Inc自增
         */
        private AtomicLong value = new AtomicLong(0);
        /**
         * 与数据库对应，maxId
         */
        private volatile long max;
        /**
         * 与数据库对应，自增范围值
         */
        private volatile int step;
        /**
         * 管理双buffer管理器
         */
        private SegmentBuffer buffer;
    }

双buffer管理器类解读：

/**
     * 双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; //当前线程是否在运行中（针对当前buffer容量达到10%后异步运行另一个buffer的扩容。）
        private final ReadWriteLock lock; //锁
    
        private volatile int step;
        private volatile int minStep;
        private volatile long updateTimestamp;
    
        public SegmentBuffer() {
            segments = new Segment[]{new Segment(this), new Segment(this)};
            currentPos = 0;
            nextReady = false;
            initOk = false;
            threadRunning = new AtomicBoolean(false);
            lock = new ReentrantReadWriteLock();
        }
        
        /**
         * 获取当前的buffer
         * @return
         */
        public Segment getCurrent() {
            return segments[currentPos];
        }
    
    
        /**
         * 读锁与写锁
         * @return
         */
        public Lock rLock() {
            return lock.readLock();
        }
    
        public Lock wLock() {
            return lock.writeLock();
        }
     }

**核心类SegmentIDGenImpl 解读：**

public class SegmentIDGenImpl implements IDGen {
        private static final Logger logger = LoggerFactory.getLogger(SegmentIDGenImpl.class);
    
        /**
         * IDCache未初始化成功时的异常码
         */
        private static final long EXCEPTION_ID_IDCACHE_INIT_FALSE = -1;
        /**
         * key不存在时的异常码
         */
        private static final long EXCEPTION_ID_KEY_NOT_EXISTS = -2;
        /**
         * SegmentBuffer中的两个Segment均未从DB中装载时的异常码
         */
        private static final long EXCEPTION_ID_TWO_SEGMENTS_ARE_NULL = -3;
        /**
         * 最大步长不超过100,0000
         */
        private static final int MAX_STEP = 1000000;
        /**
         * 一个Segment维持时间为15分钟
         */
        private static final long SEGMENT_DURATION = 15 * 60 * 1000L;
        /**
         * 用于执行定时的线程池
         */
        private ExecutorService service = new ThreadPoolExecutor(5, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>(), new UpdateThreadFactory());
        /**
         * 当前是否初始化完成
         */
        private volatile boolean initOK = false;
        /**
         * 保存不同的tag与id生成器的缓存
         */
        private Map<String, SegmentBuffer> cache = new ConcurrentHashMap<String, SegmentBuffer>();
        /**
         * 执行update，select的sql管理
         * 主要设计如下几个sql
         * getLeafAlloc：SELECT biz_tag, max_id, step FROM leaf_alloc WHERE biz_tag = #{tag}（根据tag获取当前配置参数）
         * updateMaxID：UPDATE leaf_alloc SET max_id = max_id + step WHERE biz_tag = #{tag}（根据tag更新maxId，根据当前配置的step配置）
         * updateMaxIdByCustomStep： UPDATE leaf_alloc SET max_id = max_id + #{step} WHERE biz_tag = #{key} （根据tag和指定的step更新maxId）
         *
         */
        private IDAllocDao dao;
    
        /**
         * 线程池的线程生成器
         */
        public static class UpdateThreadFactory implements ThreadFactory {
    
            private static int threadInitNumber = 0;
    
            private static synchronized int nextThreadNum() {
                return threadInitNumber++;
            }
    
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "Thread-Segment-Update-" + nextThreadNum());
            }
        }
    
        @Override
        public boolean init() {
            logger.info("Init ...");
            // 确保数据库配置的tag和配置的maxId，step等配置信息加载到kv后才初始化成功
            updateCacheFromDb();
            initOK = true;
            //定时任务更新上述的kv缓存
            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() {
                    updateCacheFromDb();
                }
            }, 60, 60, TimeUnit.SECONDS);
        }
    
        private void updateCacheFromDb() {
            logger.info("update cache from db");
            StopWatch sw = new Slf4JStopWatch();
            try {
                List<String> dbTags = dao.getAllTags();
                if (dbTags == null || dbTags.isEmpty()) {
                    return;
                }
                List<String> cacheTags = new ArrayList<String>(cache.keySet());
                List<String> insertTags = new ArrayList<String>(dbTags);
                List<String> removeTags = new ArrayList<String>(cacheTags);
                //db中新加的tags灌进cache
                insertTags.removeAll(cacheTags);
                for (String tag : insertTags) {
                    SegmentBuffer buffer = new SegmentBuffer();
                    buffer.setKey(tag);
                    Segment segment = buffer.getCurrent();
                    segment.setValue(new AtomicLong(0));
                    segment.setMax(0);
                    segment.setStep(0);
                    cache.put(tag, buffer);
                    logger.info("Add tag {} from db to IdCache, SegmentBuffer {}", tag, buffer);
                }
                //cache中已失效的tags从cache删除
                removeTags.removeAll(dbTags);
                for (String tag : removeTags) {
                    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");
            }
        }
    
        /**
         * 核心方法根据tag获取id
         * @param key
         * @return
         */
        @Override
        public Result get(final String key) {
            if (!initOK) {
                return new Result(EXCEPTION_ID_IDCACHE_INIT_FALSE, Status.EXCEPTION);
            }
            if (cache.containsKey(key)) {
                //从kv中获取双buffer管理器
                SegmentBuffer buffer = cache.get(key);
                //buffer是否初始化
                if (!buffer.isInitOk()) {
                    //加锁，并再次判断。避免初始化两次
                    synchronized (buffer) {
                        if (!buffer.isInitOk()) {
                            try {
                                //对tag和buffer的更新，即更新内存和数据库的maxId，step等
                                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);
        }
    
        public void updateSegmentFromDb(String key, Segment segment) {
            StopWatch sw = new Slf4JStopWatch();
            SegmentBuffer buffer = segment.getBuffer();
            LeafAlloc leafAlloc;
            if (!buffer.isInitOk()) {
                leafAlloc = dao.updateMaxIdAndGetLeafAlloc(key);
                buffer.setStep(leafAlloc.getStep());
                buffer.setMinStep(leafAlloc.getStep());//leafAlloc中的step为DB中的step
            } else if (buffer.getUpdateTimestamp() == 0) {
                leafAlloc = dao.updateMaxIdAndGetLeafAlloc(key);
                buffer.setUpdateTimestamp(System.currentTimeMillis());
                buffer.setStep(leafAlloc.getStep());
                buffer.setMinStep(leafAlloc.getStep());//leafAlloc中的step为DB中的step
            } else {
                long duration = System.currentTimeMillis() - buffer.getUpdateTimestamp();
                int nextStep = buffer.getStep();
                //如果15分钟内第二次进行到达超过step的10%
                if (duration < SEGMENT_DURATION) {
                    if (nextStep * 2 > MAX_STEP) {
                        //do nothing
                    } else {
                        //升级step的容量
                        nextStep = nextStep * 2;
                    }
                } else if (duration < SEGMENT_DURATION * 2) {
                    //do nothing with nextStep
                } else {
                    //如果30分钟以后才第二次进行到达超过step的10%
                    //根据当前step的大小决定是否对step容量降低
                    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);
                //更新maxid
                leafAlloc = dao.updateMaxIdByCustomStepAndGetLeafAlloc(temp);
                buffer.setUpdateTimestamp(System.currentTimeMillis());
                buffer.setStep(nextStep);
                buffer.setMinStep(leafAlloc.getStep());//leafAlloc的step为DB中的step
            }
            // must set value before set max
            long value = leafAlloc.getMaxId() - buffer.getStep();
            segment.getValue().set(value);
            segment.setMax(leafAlloc.getMaxId());
            segment.setStep(buffer.getStep());
            sw.stop("updateSegmentFromDb", key + " " + segment);
        }
    
        /**
         * 获取Id的业务逻辑，双buffer的实现逻辑
         * @param buffer
         * @return
         */
        public Result getIdFromSegmentBuffer(final SegmentBuffer buffer) {
            //重试操作，用户双buffer的切换后能正常读取id
            while (true) {
                //加锁。避免重复执行
                buffer.rLock().lock();
                try {
                    //获取当前buffer
                    final Segment segment = buffer.getCurrent();
                    //other buffer尚未准备好生成id  and  当使用使用超过10%的buffer and 当前线程当尚未运行(相当于锁，保证只会有一个线程进行这个更新操作，避免后续的id无法正常获取)
                    if (!buffer.isNextReady() && (segment.getIdle() < 0.9 * segment.getStep()) && buffer.getThreadRunning().compareAndSet(false, true)) {
                        //子线程执行数据库maxId的更新操作
                        service.execute(new Runnable() {
                            @Override
                            public void run() {
                                Segment next = buffer.getSegments()[buffer.nextPos()];
                                boolean updateOk = false;
                                try {
                                    //数据库maxId，step的更新和刷新buffer的封装方法
                                    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已经准备好
                                        buffer.setNextReady(true);
                                        //buffer更新运行完成
                                        buffer.getThreadRunning().set(false);
                                        buffer.wLock().unlock();
                                    } else {
                                        //buffer更新运行完成
                                        buffer.getThreadRunning().set(false);
                                    }
                                }
                            }
                        });
                    }
                    //获取buffer管理的value，自增后即为生成的ID
                    long value = segment.getValue().getAndIncrement();
                    //合法的ID，返回
                    if (value < segment.getMax()) {
                        return new Result(value, Status.SUCCESS);
                    }
                } finally {
                    buffer.rLock().unlock();
                }
                //等待上述的if中的子线程运行完成
                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);
                    }
                    //切换当前buffer，双buffer的最好尝试
                    if (buffer.isNextReady()) {
                        buffer.switchPos();
                        buffer.setNextReady(false);
                    } else {
                        //如果日志频繁的打印改异常，则step太小，数据消耗非常快。
                        //预留90%buffer的消耗已经使用完毕，而追加maxid的子线程还没有更新完成
                        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();
                }
            }
        }
    
        /**
         * 对于while循环中break：
         * 前面已经留了90%buffer消耗的时间来更新DB了（这部分时间已经是分钟级别的了），其实到了waitAndSleep已经可以算作是异常状态了，
         * wait 10ms是最后的尝试了。如果不进行break每次wait 10ms，wait N次，那么可能导致很多线程hang在这里。
         * @param buffer
         */
        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;
                    }
                }
            }
        }
    }

[leaf _ID_]: https://tech.meituan.com/MT_Leaf.html
[github]: https://github.com/Meituan-Dianping/Leaf
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Nhbm90_size_16_color_FFFFFF_t_70]: /images/20230531/07644fc27b3749edac461643e97c871a.png
[Ticket Servers_ Distributed Unique Primary Keys on the Cheap]: http://code.flickr.net/2010/02/08/ticket-servers-distributed-unique-primary-keys-on-the-cheap/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Nhbm90_size_16_color_FFFFFF_t_70 1]: /images/20230531/68c8897b83fa4d8ea7bdf34a369e54f9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Nhbm90_size_16_color_FFFFFF_t_70 2]: /images/20230531/7765923f3eff4066ab9d3120c41bbf9c.png