分布式系統-主鍵唯一id,訂單編號生成-雪花算法-SnowFlake

• 分布式系統下 我們每台設備(分布式系統-獨立的應用空間-或者docker環境)

  * SnowFlake的優點是，整體上按照時間自增排序，並且整個分布式系統內不會產生ID碰撞(由數據中心ID和機器ID作區分)，並且效率較高，經測試，SnowFlake每秒能夠產生26萬ID左右。

• 所以我們可以為分布式系統下:分庫分表主鍵,分庫,多庫的情況下的訂單編號使用這種方式進行唯一number操作

 

• 雖然這種方法正常情況下還是可以湊合用的,但是假如設備出現時間差,在極度大的並發情況下,還是會出現問題的,設備掩碼4095,
• 因為這個方案所支持的最小划分粒度是「毫秒 * 線程」，單線程（Snowflake 里對應的概念是 Worker）的每秒容量是12-bit，也就是接近4096.
• 當時鍾回撥則可能出現服務掛起,處於不可用狀態,有哪些時間會發生呢,比如遇到閏秒時間等等情況.

　　當然有些其他的辦法:像專門使用數據庫生成唯一id,加入需要多N台設備,每一台設備的起始值不同,步長則為N,例如:要部署N台機器，步長需設置為N，每台的初始值依次為0,1,2...N-1,

 

　　但是這種有很大缺點,一開始就要整合好:缺點如下:

• 系統水平擴展比較困難，比如定義好了步長和機器台數之后，如果要添加機器該怎么做？假設現在只有一台機器發號是1,2,3,4,5（步長是1），這個時候需要擴容機器一台。可以這樣做：把第二台機器的初始值設置得比第一台超過很多，比如14（假設在擴容時間之內第一台不可能發到14），同時設置步長為2，那么這台機器下發的號碼都是14以后的偶數。然后摘掉第一台，把ID值保留為奇數，比如7，然后修改第一台的步長為2。讓它符合我們定義的號段標准，對於這個例子來說就是讓第一台以后只能產生奇數。擴容方案看起來復雜嗎？貌似還好，現在想象一下如果我們線上有100台機器，這個時候要擴容該怎么做？簡直是噩夢。所以系統水平擴展方案復雜難以實現。
• ID沒有了單調遞增的特性，只能趨勢遞增，這個缺點對於一般業務需求不是很重要，可以容忍。
• 數據庫壓力還是很大，每次獲取ID都得讀寫一次數據庫，只能靠堆機器來提高性能。

　　

　　Leaf方案實現

　　Leaf這個名字是來自德國哲學家、數學家萊布尼茨的一句話：There are no two identical leaves in the world　"世界上沒有兩片相同的樹葉"

　　綜合對比上述幾種方案，每種方案都不完全符合我們的要求。所以Leaf分別在上述第二種和第三種方案上做了相應的優化，實現了Leaf-segment和Leaf-snowflake方案。

　　1.Leaf-segment數據庫方案 2. Leaf-snowflake方案  具體可參考搜索. 也可去搜索一下 zookeeper的分布式唯一

 

 

 

Twitter_Snowflake 介紹:
* SnowFlake的結構如下(每部分用-分開):
* 0 - 0000000000 0000000000 0000000000 0000000000 0 - 00000 - 00000 - 000000000000
* 1位標識，由於long基本類型在Java中是帶符號的，最高位是符號位，正數是0，負數是1，所以id一般是正數，最高位是0
* 41位時間截(毫秒級)，注意，41位時間截不是存儲當前時間的時間截，而是存儲時間截的差值（當前時間截 - 開始時間截)得到的值），
* 這里的的開始時間截，一般是我們的id生成器開始使用的時間，由我們程序來指定的（如下下面程序IdWorker類的startTime屬性）。41位的時間截，可以使用69年，年T = (1L << 41) / (1000L * 60 * 60 * 24 * 365) = 69
* 10位的數據機器位，可以部署在1024個節點，包括5位datacenterId和5位workerId,12位序列，毫秒內的計數，12位的計數順序號支持每個節點每毫秒(同一機器，同一時間截)產生4096個ID序號加起來剛好64位，為一個Long型。
  
唯一的缺點是設備時間要穩定,不能出現回退.

下面就是我的代碼: 沒有代碼不就是白瞎了?

package test;

/**
 * @Title: SnowFlakeUtils.java
 * @Package com.cn.alasga.common.core.util.wechat
 * @Description: 雪花算法生成不重復的序列號 可用作訂單編號
 * @author LiJing
 * @date 2018/12/6 13:46
 * @version v.3.0
 */

import org.apache.curator.shaded.com.google.common.util.concurrent.ThreadFactoryBuilder;

import java.util.concurrent.*;

/**
 * Twitter_Snowflake<br>
 * SnowFlake的結構如下(每部分用-分開):<br>
 * 0 - 0000000000 0000000000 0000000000 0000000000 0 - 00000 - 00000 - 000000000000 <br>
 * 1位標識，由於long基本類型在Java中是帶符號的，最高位是符號位，正數是0，負數是1，所以id一般是正數，最高位是0<br>
 * 41位時間截(毫秒級)，注意，41位時間截不是存儲當前時間的時間截，而是存儲時間截的差值（當前時間截 - 開始時間截)
 * 得到的值），這里的的開始時間截，一般是我們的id生成器開始使用的時間，由我們程序來指定的（如下下面程序IdWorker類的startTime屬性）。41位的時間截，可以使用69年，年T = (1L << 41) / (1000L * 60 * 60 * 24 * 365) = 69<br>
 * 10位的數據機器位，可以部署在1024個節點，包括5位datacenterId和5位workerId<br>
 * 12位序列，毫秒內的計數，12位的計數順序號支持每個節點每毫秒(同一機器，同一時間截)產生4096個ID序號<br>
 * 加起來剛好64位，為一個Long型。<br>
 * SnowFlake的優點是，整體上按照時間自增排序，並且整個分布式系統內不會產生ID碰撞(由數據中心ID和機器ID作區分)，並且效率較高，經測試，SnowFlake每秒能夠產生26萬ID左右。
 */
public class SnowflakeUtils {

    private static SnowflakeUtils idGenerater;

    static {
        idGenerater = new SnowflakeUtils(1, 2);
    }

    public synchronized static long getOrderNo() {
        return idGenerater.nextId();
    }

    // ==============================Fields===========================================
    /**
     * 開始時間截 (2015-01-01)
     */
    private final long twepoch = 1420041600000L;

    /**
     * 機器id所占的位數
     */
    private final long workerIdBits = 5L;

    /**
     * 數據標識id所占的位數
     */
    private final long datacenterIdBits = 5L;

    /**
     * 支持的最大機器id，結果是31 (這個移位算法可以很快的計算出幾位二進制數所能表示的最大十進制數)
     */
    private final long maxWorkerId = -1L ^ (-1L << workerIdBits);

    /**
     * 支持的最大數據標識id，結果是31
     */
    private final long maxDatacenterId = -1L ^ (-1L << datacenterIdBits);

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

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

    /**
     * 數據標識id向左移17位(12+5)
     */
    private final long datacenterIdShift = sequenceBits + workerIdBits;

    /**
     * 時間截向左移22位(5+5+12)
     */
    private final long timestampLeftShift = sequenceBits + workerIdBits + datacenterIdBits;

    /**
     * 生成序列的掩碼，這里為4095 (0b111111111111=0xfff=4095)
     */
    private final long sequenceMask = -1L ^ (-1L << sequenceBits);

    /**
     * 工作機器ID(0~31)
     */
    private long workerId;

    /**
     * 數據中心ID(0~31)
     */
    private long datacenterId;

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

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

    //==============================Constructors=====================================

    /**
     * 構造函數
     *
     * @param workerId     工作ID (0~31)
     * @param datacenterId 數據中心ID (0~31)
     */
    public SnowflakeUtils(long workerId, long datacenterId) {
        if (workerId > maxWorkerId || workerId < 0) {
            throw new IllegalArgumentException(String.format("worker Id can't be greater than %d or less than 0", maxWorkerId));
        }
        if (datacenterId > maxDatacenterId || datacenterId < 0) {
            throw new IllegalArgumentException(String.format("datacenter Id can't be greater than %d or less than 0", maxDatacenterId));
        }
        this.workerId = workerId;
        this.datacenterId = datacenterId;
    }

    // ==============================Methods==========================================

    /**
     * 獲得下一個ID (該方法是可以加注 synchronized 線程安全)
     *
     * @return SnowflakeId
     */
    private 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) {
            sequence = (sequence + 1) & sequenceMask;
            //毫秒內序列溢出
            if (sequence == 0) {
                //阻塞到下一個毫秒,獲得新的時間戳
                timestamp = tilNextMillis(lastTimestamp);
            }
        }
        //時間戳改變，毫秒內序列重置
        else {
            sequence = 0L;
        }

        //上次生成ID的時間截
        lastTimestamp = timestamp;

        //移位並通過或運算拼到一起組成64位的ID
        return ((timestamp - twepoch) << timestampLeftShift)
                | (datacenterId << datacenterIdShift)
                | (workerId << workerIdShift)
                | sequence;
    }

    /**
     * 阻塞到下一個毫秒，直到獲得新的時間戳
     *
     * @param lastTimestamp 上次生成ID的時間截
     * @return 當前時間戳
     */
    protected long tilNextMillis(long lastTimestamp) {
        long timestamp = timeGen();
        while (timestamp <= lastTimestamp) {
            timestamp = timeGen();
        }
        return timestamp;
    }

    /**
     * 返回以毫秒為單位的當前時間
     *
     * @return 當前時間(毫秒)
     */
    protected long timeGen() {
        return System.currentTimeMillis();
    }

    //==============================Test=============================================

    /**
     * 測試
     */
    public static void main(String[] args) throws Exception {
        // 線程數量
        final int threadCount = 100;
        // 每個線程生成的 ID 數量
        final int idCountPerThread = 1000;
        // 用於等待所有線程啟動完成
        CountDownLatch threadLatch = new CountDownLatch(threadCount);

        final int coreThread = 5;
        final int maxThread = 50;
        final long keepAliveTime = 0L;
        final int queueCapacity = 1024;


        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("demo-pool-%d").build();

        //Common Thread Pool
        ExecutorService pool = new ThreadPoolExecutor(coreThread, maxThread, keepAliveTime, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>(queueCapacity), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());


        ConcurrentSkipListSet<Long> ids = new ConcurrentSkipListSet<>();
        for (int i = 0; i < threadCount; ++i) {
            final int n = i;
            pool.execute(() -> {
                // 等待所有線程都運行到這里，然后都繼續運行，差不多同時生成 id
                final String threadNum = Thread.currentThread().getName() + "-" + n + "號線程";
                try {
                    threadLatch.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
//                System.out.println(threadNum+"繼續執行");
                for (int j = 0; j < idCountPerThread; ++j) {
                    long id = SnowflakeUtils.getOrderNo();
                    ids.add(id);
                    System.out.println(id);
                }
            });
            threadLatch.countDown();
        }
        pool.shutdown();
        // 等待 id 生成完成，生成不同數量的 id 時需要調整
        Thread.sleep(2000);
//        System.out.println(ids.size());
        //System.out.println(ids);
    }
}

　　

現在來解釋一下:

 

代碼中測試用到的 ConcurrentSkipListSet<Long> idListSet = new ConcurrentSkipListSet<>(); ConcurrentSkipListSet是線程安全的有序的集合，適用於高並發的場景。 ConcurrentSkipListSet和TreeSet，它們雖然都是有序的集合。 但是，第一，它們的線程安全機制不同，TreeSet是非線程安全的，而ConcurrentSkipListSet是線程安全的。 第二，ConcurrentSkipListSet是通過ConcurrentSkipListMap實現的，而TreeSet是通過TreeMap實現的。
說明:
(01) ConcurrentSkipListSet繼承於AbstractSet。因此，它本質上是一個集合。
(02) ConcurrentSkipListSet實現了NavigableSet接口。因此，ConcurrentSkipListSet是一個有序的集合。
(03) ConcurrentSkipListSet是通過ConcurrentSkipListMap實現的。它包含一個ConcurrentNavigableMap對象m，而m對象實際上是ConcurrentNavigableMap的實現類ConcurrentSkipListMap的實例。ConcurrentSkipListMap中的元素是key-value鍵值對；而ConcurrentSkipListSet是集合，它只用到了ConcurrentSkipListMap中的key！
(04) 里面不能有重復的數據,而且是有序的,還支持並發,很強大啊!!

但是::::

查看add()方法的javadoc，其注釋為：
如果此 set 中不包含指定元素，則添加指定元素。更確切地講，如果此 set 不包含滿足 e.equals(e2) 的元素 
e2，則向 set 中添加指定的元素 e。如果此 set 已經包含該元素，則調用不更改該 set 並返回 
false。


根據注釋，只要元素的equals()方法判斷不相等就能加入到Set中，可調試發現不是這么回事：

Java代碼  

public boolean add(E e) {
    return m.putIfAbsent(e, Boolean.TRUE) == null;
}

m為 ConcurrentSkipListMap，它的putIfAbsent()方法實現是：

Java代碼  

　　

    public V putIfAbsent(K key, V value) {
        if (value == null)
            throw new NullPointerException();
        return doPut(key, value, true);
    }

    private V doPut(K kkey, V value, boolean onlyIfAbsent) {
        Comparable super K> key = comparable(kkey);
        for (;;) {
            Node b = findPredecessor(key);
            Node n = b.next;
            for (;;) {
                if (n != null) {
                    Node f = n.next;
                    if (n != b.next)               // inconsistent read
                        break;;
                    Object v = n.value;
                    if (v == null) {               // n is deleted
                        n.helpDelete(b, f);
                        break;
                    }
                    if (v == n || b.value == null) // b is deleted
                        break;
                    int c = key.compareTo(n.key);
                    if (c > 0) {
                        b = n;
                        n = f;
                        continue;
                    }
                    if (c == 0) {
                        if (onlyIfAbsent || n.casValue(v, value))
                            return (V)v;
                        else
                            break; // restart if lost race to replace value
                    }
                    // else c  z = new Node(kkey, value, n);
                if (!b.casNext(n, z))
                    break;         // restart if lost race to append to b
                int level = randomLevel();
                if (level > 0)
                    insertIndex(z, level);
                return null;
            }
        }
    }

 

  實際上Set中的添加是根據元素的compareTo()方法在比較！如果compareTo()返回0，就認為2個元素相等，跟equals()方法根本沒有關系！JAVA SDK 的JavaDoc也不靠譜啊~~