分布式UID解决方案分析

CREATE TABLE id_generator (
  id int(10) NOT NULL,
  max_id bigint(20) NOT NULL COMMENT '当前最大的可用id',
  step int(20) NOT NULL COMMENT '号段的长度',
  biz_id	int(20) NOT NULL COMMENT '业务id',
  version int(20) NOT NULL COMMENT '版本号',
  PRIMARY KEY (`id`)
) 

update id_generator set max_id = max_id+step, version = version + 1 where version = version and biz_id = ***

CREATE TABLE WORKER_NODE(
ID BIGINT NOT NULL AUTO_INCREMENT COMMENT 'auto increment id',
HOST_NAME VARCHAR(64) NOT NULL COMMENT 'host name',
PORT VARCHAR(64) NOT NULL COMMENT 'port',
TYPE INT NOT NULL COMMENT 'node type: ACTUAL or CONTAINER',
LAUNCH_DATE DATE NOT NULL COMMENT 'launch date',
MODIFIED TIMESTAMP NOT NULL COMMENT 'modified time',
CREATED TIMESTAMP NOT NULL COMMENT 'created time',
PRIMARY KEY(ID)
) COMMENT='DB WorkerID Assigner for UID Generator',ENGINE = INNODB;

public interface UidGenerator {

    /**
     * 获取UID
     */
    long getUID() throws UidGenerateException;

    /**
     * 解析格式化输出UID
     */
    String parseUID(long uid);
}

 protected synchronized long nextId() {
     	//获取当前时间，单位秒，会校验是否超过最大的时间年限，超过则抛出异常
        long currentSecond = getCurrentSecond();

        // 如果出现时间回拨，就抛出异常
        if (currentSecond < lastSecond) {
            long refusedSeconds = lastSecond - currentSecond;
            throw new UidGenerateException("Clock moved backwards. Refusing for %d seconds", refusedSeconds);
        }

        // 同一秒内序列号自增，
        if (currentSecond == lastSecond) {
            sequence = (sequence + 1) & bitsAllocator.getMaxSequence();
            //超过最大序列，等待下一秒生成uid
            if (sequence == 0) {
                currentSecond = getNextSecond(lastSecond);
            }

        // 不同秒内序列号从0开始
        } else {
            sequence = 0L;
        }

        lastSecond = currentSecond;

        // bits位移操作生成UID
        return bitsAllocator.allocate(currentSecond - epochSeconds, workerId, sequence);
    }

	<!-- 以下为可选配置, 如未指定将采用默认值 -->
		<!-- RingBuffer size扩容参数, 可提高UID生成的吞吐量. --> 
		<!-- 默认:3， 原bufferSize=8192, 扩容后bufferSize= 8192 << 3 = 65536 -->
		<property name="boostPower" value="3"></property> 
		
		<!-- 指定何时向RingBuffer中填充UID, 取值为百分比(0, 100), 默认为50 -->
		<!-- 举例: bufferSize=1024, paddingFactor=50 -> threshold=1024 * 50 / 100 = 512. -->
		<!-- 当环上可用UID数量 < 512时, 将自动对RingBuffer进行填充补全 -->
		<property name="paddingFactor" value="50"></property> 
		
		<!-- 另外一种RingBuffer填充时机, 在Schedule线程中, 周期性检查填充 -->
		<!-- 默认:不配置此项, 即不使用Schedule线程. 如需使用, 请指定Schedule线程时间间隔, 单位:秒 -->
		<property name="scheduleInterval" value="60"></property> 
		
		<!-- 拒绝策略: 当环已满, 无法继续填充时 -->
		<!-- 默认无需指定, 将丢弃Put操作, 仅日志记录. 如有特殊需求, 请实现RejectedPutBufferHandler接口(支持Lambda表达式) -->
		<property name="rejectedPutBufferHandler" ref="XxxxYourPutRejectPolicy"></property> 
		
		<!-- 拒绝策略: 当环已空, 无法继续获取时 -->
		<!-- 默认无需指定, 将记录日志, 并抛出UidGenerateException异常. 如有特殊需求, 请实现RejectedTakeBufferHandler接口(支持Lambda表达式) -->
		<property name="rejectedTakeBufferHandler" ref="XxxxYourTakeRejectPolicy"></property>

public class PaddedAtomicLong extends AtomicLong {
    private static final long serialVersionUID = -3415778863941386253L;

    /** Padded 6 long (48 bytes) */
    public volatile long p1, p2, p3, p4, p5, p6 = 7L;

    /**
     * Constructors from {@link AtomicLong}
     */
    public PaddedAtomicLong() {
        super();
    }

    public PaddedAtomicLong(long initialValue) {
        super(initialValue);
    }

    /**
     * To prevent GC optimizations for cleaning unused padded references
     */
    public long sumPaddingToPreventOptimization() {
        return p1 + p2 + p3 + p4 + p5 + p6;
    }

}

public void paddingBuffer() {
        LOGGER.info("Ready to padding buffer lastSecond:{}. {}", lastSecond.get(), ringBuffer);

        // is still running
        if (!running.compareAndSet(false, true)) {
            LOGGER.info("Padding buffer is still running. {}", ringBuffer);
            return;
        }
    
        boolean isFullRingBuffer = false;
        while (!isFullRingBuffer) {
            //调用nextIdsForOneSecond方法获取UID list，注意此处的时间使用lastSecond.incrementAndGet()方法时间自增的方式获取而不是System.currentTimeMillis()获取，解决了运行时时间回拨问题
            List<Long> uidList = uidProvider.provide(lastSecond.incrementAndGet());
            for (Long uid : uidList) {
                //调用put方法加入RingBuffer of UID中
                isFullRingBuffer = !ringBuffer.put(uid);
                if (isFullRingBuffer) {
                    break;
                }
            }
        }

        // not running now
        running.compareAndSet(true, false);
        LOGGER.info("End to padding buffer lastSecond:{}. {}", lastSecond.get(), ringBuffer);
    }

/**
     * 指定秒内获取UID List，Size为max sequence（0~max sequence）
     * 
     * @param currentSecond
     * @return UID list, size of {@link BitsAllocator#getMaxSequence()} + 1
     */
    protected List<Long> nextIdsForOneSecond(long currentSecond) {
        // Initialize result list size of (max sequence + 1)
        int listSize = (int) bitsAllocator.getMaxSequence() + 1;
        List<Long> uidList = new ArrayList<>(listSize);

        // Allocate the first sequence of the second, the others can be calculated with the offset
        long firstSeqUid = bitsAllocator.allocate(currentSecond - epochSeconds, workerId, 0L);
        for (int offset = 0; offset < listSize; offset++) {
            uidList.add(firstSeqUid + offset);
        }

        return uidList;
    }

public synchronized boolean put(long uid) {
    	//获取Tail与Cursor指针
        long currentTail = tail.get();
        long currentCursor = cursor.get();

        // 1. 判断Ring是否已满，满了执行Put拒绝策略
        long distance = currentTail - (currentCursor == START_POINT ? 0 : currentCursor);
        if (distance == bufferSize - 1) {
            rejectedPutHandler.rejectPutBuffer(this, uid);
            return false;
        }

        // 2. check标志是否为CAN_PUT_FLAG
        int nextTailIndex = calSlotIndex(currentTail + 1);
        if (flags[nextTailIndex].get() != CAN_PUT_FLAG) {
            rejectedPutHandler.rejectPutBuffer(this, uid);
            return false;
        }

        // 3. 将UID插入下一个slots中
        // 4. 设置标志为CAN_TAKE_FLAG
        // 5. tail指针+1
        slots[nextTailIndex] = uid;
        flags[nextTailIndex].set(CAN_TAKE_FLAG);
        tail.incrementAndGet();
        return true;
    }

public long take() {
        // 获取当前Cursor指针与下一个可用Cursor指针并进行check
        long currentCursor = cursor.get();
        long nextCursor = cursor.updateAndGet(old -> old == tail.get() ? old : old + 1);
        Assert.isTrue(nextCursor >= currentCursor, "Curosr can't move back");

        // 1. 判断如果达到阈值，则以异步模式触发填充
        long currentTail = tail.get();
        if (currentTail - nextCursor < paddingThreshold) {
            LOGGER.info("Reach the padding threshold:{}. tail:{}, cursor:{}, rest:{}", paddingThreshold, currentTail,
                    nextCursor, currentTail - nextCursor);
            bufferPaddingExecutor.asyncPadding();
        }

        // 2. 当前Cursor指针等于下一个可用Cursor指针，Ring为空，执行Take拒绝策略
        if (nextCursor == currentCursor) {
            rejectedTakeHandler.rejectTakeBuffer(this);
        }

        // 3. 检查下一个slot是否为CAN_TAKE_FLAG
        int nextCursorIndex = calSlotIndex(nextCursor);
        Assert.isTrue(flags[nextCursorIndex].get() == CAN_TAKE_FLAG, "Curosr not in can take status");

        // 4. 从slot中获取UID
        // 5. 设置标志为CAN_PUT_FLAG.
        long uid = slots[nextCursorIndex];
        flags[nextCursorIndex].set(CAN_PUT_FLAG);
        return uid;
    }

CREATE TABLE `leaf_alloc` (
  `biz_tag` varchar(128) NOT NULL DEFAULT '' COMMENT '业务key',
  `max_id` bigint(20) NOT NULL DEFAULT '1' COMMENT '当前已经分配了的最大id',
  `step` int(11) NOT NULL COMMENT '初始步长，也是动态调整的最小步长',
  `description` varchar(256) DEFAULT NULL COMMENT '业务key的描述',
  `update_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '数据库维护的更新时间',
  PRIMARY KEY (`biz_tag`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

@Override
    public boolean init() {
        //同步biz_tag到cache中
        updateCacheFromDb();
        initOK = true;
        //开启定时任务定时调度updateCacheFromDb()
        updateCacheFromDbAtEveryMinute();
        return initOK;
    }

private void updateCacheFromDb() {
    		//获取数据库表中所有的biz_tag
            List<String> dbTags = dao.getAllTags();
            if (dbTags == null || dbTags.isEmpty()) {
                return;
            }
    		//缓存的biz_tag
            List<String> cacheTags = new ArrayList<String>(cache.keySet());
    		Set<String> insertTagsSet = new HashSet<>(dbTags);
            Set<String> removeTagsSet = new HashSet<>(cacheTags);
            //将db中新加的biz_tags加入cache中，并初始化SegmentBuffer
            for(int i = 0; i < cacheTags.size(); i++){
                String tmp = cacheTags.get(i);
                if(insertTagsSet.contains(tmp)){
                    insertTagsSet.remove(tmp);
                }
            }
            for (String tag : insertTagsSet) {
                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);
            }
            //将db中被删除的biz_tag(已失效)从cache删除
            for(int i = 0; i < dbTags.size(); i++){
                String tmp = dbTags.get(i);
                if(removeTagsSet.contains(tmp)){
                    removeTagsSet.remove(tmp);
                }
            }
            for (String tag : removeTagsSet) {
                cache.remove(tag);
            }
      
    }

public class SegmentBuffer {
    private String key;// 数据库的biz_tag
    private Segment[] segments; //双buffer
    private volatile int currentPos; //当前的使用的segment的index
    private volatile boolean nextReady; //下一个segment是否处于可切换状态
    private volatile boolean initOk; //是否初始化完成
    private final AtomicBoolean threadRunning; //线程是否在运行中
    private final ReadWriteLock lock;

    private volatile int step;// 动态调整的step
    private volatile int minStep; // 最小step
    private volatile long updateTimestamp;// 更新时间戳
}

public class Segment {
    private AtomicLong value = new AtomicLong(0);
    private volatile long max;
    private volatile int step;
    private SegmentBuffer buffer;
}

    public Result get(final String key) {
        //SegmentIDGenImpl是否初始化
        ...
        //cache中是否已经缓存了对应biz_id的数据
        if (cache.containsKey(key)) {
            //检查SegmentBuffer中的Segment是否初始化
                    if (!buffer.isInitOk()) {
                            //初始化Segment
                            updateSegmentFromDb(key, buffer.getCurrent());
                            buffer.setInitOk(true);
                }
        }
            return getIdFromSegmentBuffer(cache.get(key));
}

    public void updateSegmentFromDb(String key, Segment segment) {
        StopWatch sw = new Slf4JStopWatch();
        SegmentBuffer buffer = segment.getBuffer();
        LeafAlloc leafAlloc;
        /**
         * 第一次初始号段
         * 1. 更新数据库中key对应记录的maxId（maxId=maxId+step)
         * 2. 设置buffer的Step(步长)与MinStep
         */
        if (!buffer.isInitOk()) {
            leafAlloc = dao.updateMaxIdAndGetLeafAlloc(key);
            buffer.setStep(leafAlloc.getStep());
            buffer.setMinStep(leafAlloc.getStep());//leafAlloc中的step为DB中的step
        }
        /**
         * 第二次初始号段（双Buffer异步准备）
         * 1. 更新数据库中key对应记录的maxId（maxId=maxId+step)
         * 2. 设置buffer的Step(步长)与MinStep
         */
        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
        }
        /**
         * 后续双Buffer异步准备
         * 动态调整step
         *      1. duration < 15 分钟 : step 变为原来的2倍， 最大为 MAX_STEP
         *      2. 15分钟 <= duration < 30分钟 : nothing
         *      3. duration >= 30 分钟 : 缩小step, 最小为DB中配置的step
         * 更新数据库中key对应记录的maxId（maxId=maxId+step)
         * 设置buffer的Step(步长)与MinStep
         */
        else {
            //更新时间差
            long duration = System.currentTimeMillis() - buffer.getUpdateTimestamp();
            int nextStep = buffer.getStep();
            if (duration < SEGMENT_DURATION) {
                if (nextStep * 2 > MAX_STEP) {
                    //do nothing
                } else {
                    nextStep = nextStep * 2;
                }
            } else if (duration < SEGMENT_DURATION * 2) {
                //do nothing with nextStep
            } else {
                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);
            leafAlloc = dao.updateMaxIdByCustomStepAndGetLeafAlloc(temp);
            buffer.setUpdateTimestamp(System.currentTimeMillis());
            buffer.setStep(nextStep);
            buffer.setMinStep(leafAlloc.getStep());//leafAlloc的step为DB中的step
        }

        /**
         * 准备当前Segment号段，设置value初始值与max/step值为buffer的max与step
         */
        long value = leafAlloc.getMaxId() - buffer.getStep();
        segment.getValue().set(value);
        segment.setMax(leafAlloc.getMaxId());
        segment.setStep(buffer.getStep());
        sw.stop("updateSegmentFromDb", key + " " + segment);
    }

//发生了回拨，此刻时间小于上次发号时间
 if (timestamp < lastTimestamp) {
  			  
            long offset = lastTimestamp - timestamp;
            if (offset <= 5) {
                try {
                	//时间偏差大小小于5ms，则等待两倍时间
                    wait(offset << 1);//wait
                    timestamp = timeGen();
                    if (timestamp < lastTimestamp) {
                       //还是小于，抛异常并上报
                        throwClockBackwardsEx(timestamp);
                      }    
                } catch (InterruptedException e) {  
                   throw  e;
                }
            } else {
                //throw
                throwClockBackwardsEx(timestamp);
            }
        }