本文主要介紹 uid-generator （一種全局ID服務(wù)實(shí)現(xiàn)）

uid-generator介紹

全局ID服務(wù)是分布式服務(wù)中的基礎(chǔ)服務(wù)，需要保持全局唯一，高效，高可靠性。有些時(shí)候還可能要求保持單調(diào)，但也并非一定要嚴(yán)格遞增或者遞減

全局ID也可以通過(guò)數(shù)據(jù)庫(kù)的自增主鍵來(lái)獲取，但是如果要求QPS很高顯然是不現(xiàn)實(shí)的

uid-generator是對(duì)Snowflake算法的改進(jìn)，也引入了高性能隊(duì)列disruptor中RingBuffer思想，進(jìn)一步提升了效率

  +------+----------------------+----------------+-----------+
  | sign |     delta seconds    | worker node id | sequence  |
  +------+----------------------+----------------+-----------+
    1bit          28bits              22bits         13bits

• sign 符號(hào)位 保證為正數(shù)

• delta seconds 當(dāng)前時(shí)間與約定時(shí)間的差值

• word node id機(jī)器碼

• sequence 同一時(shí)刻支持并發(fā)數(shù)

上圖就是snowflake算法生成的64位的長(zhǎng)整型構(gòu)成

uid-generator的work node id 使用了數(shù)據(jù)庫(kù)自增主鍵的key，每次重啟服務(wù)都需要刷新，這也保證了集群中全局ID的唯一性

worker node id字段處理

uid-generator使用數(shù)據(jù)庫(kù)主鍵作為worker node id

這樣看來(lái)這個(gè)worker node id其實(shí)可以有很豐富的擴(kuò)展性，只要對(duì)表結(jié)構(gòu)稍微修改，就可以記錄使得worker node id 有更有意義的含義。

例如使用uid-generator生成的值作為表的主鍵ID，可以通過(guò)對(duì)WORKER_NODE表增加一列表名記錄表，這樣通過(guò)id就反向查找對(duì)應(yīng)的表名

sequence字段的處理

uid-generator中實(shí)現(xiàn)了原生的snowflake以及緩存版的。這兩個(gè)版本對(duì)于sequence字段的處理有所不同

DefaultUidGenerator.java

/**
* Get UID
*
* @return UID
* @throws UidGenerateException in the case: Clock moved backwards; Exceeds the max timestamp
*/
protected synchronized long nextId() {
    long currentSecond = getCurrentSecond();

    // Clock moved backwards, refuse to generate uid
    if (currentSecond < lastSecond) {
        long refusedSeconds = lastSecond - currentSecond;
        throw new UidGenerateException("Clock moved backwards. Refusing for %d seconds", refusedSeconds);
    }

    // At the same second, increase sequence
    if (currentSecond == lastSecond) {
        sequence = (sequence + 1) & bitsAllocator.getMaxSequence();
        // Exceed the max sequence, we wait the next second to generate uid
        if (sequence == 0) {
            currentSecond = getNextSecond(lastSecond);
        }

    // At the different second, sequence restart from zero
    } else {
        sequence = 0L;
    }

    lastSecond = currentSecond;

    // Allocate bits for UID
    return bitsAllocator.allocate(currentSecond - epochSeconds, workerId, sequence);
}

DefaultUidGenerator 并發(fā)通過(guò) 函數(shù)加鎖控制；獲取seq時(shí)通過(guò)時(shí)間判斷是否需要調(diào)到下一秒

CachedUidGenerator.java

 /**
    * Padding buffer fill the slots until to catch the cursor
    */
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;
    }

    // fill the rest slots until to catch the cursor
    boolean isFullRingBuffer = false;
    while (!isFullRingBuffer) {
        List<Long> uidList = uidProvider.provide(lastSecond.incrementAndGet());
        for (Long uid : uidList) {
            isFullRingBuffer = !ringBuffer.put(uid);
            if (isFullRingBuffer) {
                break;
            }
        }
    }

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

CachedUidGenerator 加鎖通過(guò)CAS操作；由于需要一次填充完緩存，所以選取了一次填充一秒內(nèi)所有的seq，以此保證了seq在一秒內(nèi)的唯一性。這樣帶來(lái)的一個(gè)小弊端是不能通過(guò)id看出來(lái)這個(gè)id生成的時(shí)間

CachedUidGenerator核心RingBuffer實(shí)現(xiàn)

RingBuffer是一個(gè)環(huán)形數(shù)組，通過(guò)兩個(gè)指針，tail、cursor來(lái)實(shí)現(xiàn)復(fù)用槽

在這里需要介紹一下FalseShare陷阱，由于tail和cursor指針在高并發(fā)情況下變動(dòng)頻繁，如果tail，cursor處于同一個(gè)緩存中，將會(huì)頻繁導(dǎo)致緩存失效，可以看到uid-generator已經(jīng)考慮了這個(gè)問(wèn)題

通過(guò)對(duì)PaddedAtomicLong進(jìn)行填充，保證了每一個(gè)long值都在不同的緩存行中，解決了這個(gè)問(wèn)題

RingBuffer基本都用位運(yùn)算取代了乘除以及取模計(jì)算，提高了計(jì)算效率

/**
    * Put an UID in the ring & tail moved<br>
    * We use 'synchronized' to guarantee the UID fill in slot & publish new tail sequence as atomic operations<br>
    * 
    * <b>Note that: </b> It is recommended to put UID in a serialize way, cause we once batch generate a series UIDs and put
    * the one by one into the buffer, so it is unnecessary put in multi-threads
    *
    * @param uid
    * @return false means that the buffer is full, apply {@link RejectedPutBufferHandler}
    */
public synchronized boolean put(long uid) {
    long currentTail = tail.get();
    long currentCursor = cursor.get();

    // tail catches the cursor, means that you can't put any cause of RingBuffer is full
    long distance = currentTail - (currentCursor == START_POINT ? 0 : currentCursor);
    if (distance == bufferSize - 1) {
        rejectedPutHandler.rejectPutBuffer(this, uid);
        return false;
    }

    // 1. pre-check whether the flag is CAN_PUT_FLAG
    int nextTailIndex = calSlotIndex(currentTail + 1);
    if (flags[nextTailIndex].get() != CAN_PUT_FLAG) {
        rejectedPutHandler.rejectPutBuffer(this, uid);
        return false;
    }

    // 2. put UID in the next slot
    // 3. update next slot' flag to CAN_TAKE_FLAG
    // 4. publish tail with sequence increase by one
    slots[nextTailIndex] = uid;
    flags[nextTailIndex].set(CAN_TAKE_FLAG);
    tail.incrementAndGet();

    // The atomicity of operations above, guarantees by 'synchronized'. In another word,
    // the take operation can't consume the UID we just put, until the tail is published(tail.incrementAndGet())
    return true;
}

在RingBuffer的put方法中可以看到整個(gè)的流程，首先是函數(shù)加鎖，加鎖的原因在注釋部分也解釋了，由于是每次批量存入的，多線(xiàn)程put操作是沒(méi)有必要的，之后第一步計(jì)算tail與cursor距離當(dāng)前數(shù)組是否還可以繼續(xù)填充。這里還有另外一個(gè)標(biāo)識(shí)位用來(lái)判斷當(dāng)前槽是否可以做PUT以及TAKE操作，更像是雙保險(xiǎn)，防止上一個(gè)判斷距離結(jié)束了之后tail位置有變動(dòng)，導(dǎo)致槽位被覆蓋

同樣對(duì)于take操作

    /**
     * Take an UID of the ring at the next cursor, this is a lock free operation by using atomic cursor<p>
     * 
     * Before getting the UID, we also check whether reach the padding threshold, 
     * the padding buffer operation will be triggered in another thread<br>
     * If there is no more available UID to be taken, the specified {@link RejectedTakeBufferHandler} will be applied<br>
     * 
     * @return UID
     * @throws IllegalStateException if the cursor moved back
     */
    public long take() {
        // spin get next available cursor
        long currentCursor = cursor.get();
        long nextCursor = cursor.updateAndGet(old -> old == tail.get() ? old : old + 1);

        // check for safety consideration, it never occurs
        Assert.isTrue(nextCursor >= currentCursor, "Curosr can't move back");

        // trigger padding in an async-mode if reach the threshold
        long currentTail = tail.get();
        if (currentTail - nextCursor < paddingThreshold) {
            LOGGER.info("Reach the padding threshold:{}. tail:{}, cursor:{}, rest:{}", paddingThreshold, currentTail,
                    nextCursor, currentTail - nextCursor);
            bufferPaddingExecutor.asyncPadding();
        }

        // cursor catch the tail, means that there is no more available UID to take
        if (nextCursor == currentCursor) {
            rejectedTakeHandler.rejectTakeBuffer(this);
        }

        // 1. check next slot flag is CAN_TAKE_FLAG
        int nextCursorIndex = calSlotIndex(nextCursor);
        Assert.isTrue(flags[nextCursorIndex].get() == CAN_TAKE_FLAG, "Curosr not in can take status");

        // 2. get UID from next slot
        // 3. set next slot flag as CAN_PUT_FLAG.
        long uid = slots[nextCursorIndex];
        flags[nextCursorIndex].set(CAN_PUT_FLAG);

        // Note that: Step 2,3 can not swap. If we set flag before get value of slot, the producer may overwrite the
        // slot with a new UID, and this may cause the consumer take the UID twice after walk a round the ring
        return uid;
    }

正如注釋中所說(shuō)，take部分并沒(méi)有并發(fā)限制，在剩余可用槽位小于一個(gè)閾值的時(shí)候，會(huì)觸發(fā)一次填充操作

CachedUidGenerator 對(duì)于填充有兩種處理，一個(gè)是低于閾值填充，一種是開(kāi)啟Schedule，定時(shí)填充，定時(shí)填充可選

uid-generator可靠性很高，除了workid依賴(lài)數(shù)據(jù)庫(kù)之外基本不依賴(lài)外部中間件，全局ID在分布式服務(wù)中扮演關(guān)鍵角色，一旦服務(wù)出錯(cuò)，解決起來(lái)也很棘手。