【分布式技术专题】「分布式ID系列」百度开源的分布式高性能的唯一ID生成器UidGenerator

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);
    }
复制代码

@Override
    public long getUID() {
        try {
            return ringBuffer.take();
        } catch (Exception e) {
            LOGGER.error("Generate unique id exception. ", e);
            throw new UidGenerateException(e);
        }
    }
复制代码

/**
     * 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) {
            //获取生成的id，放到RingBuffer中。
            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);
    }
复制代码

/**
     * Get the UIDs in the same specified second under the 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
        //这里的实现很取巧
        //因为1秒内生成的id是连续的，所以利用第1个id来生成后面的id，而不用频繁调用snowflake算法
        long firstSeqUid = bitsAllocator.allocate(currentSecond - epochSeconds, workerId, 0L);
        for (int offset = 0; offset < listSize; offset++) {
            uidList.add(firstSeqUid + offset);
        }
        return uidList;
    }
复制代码

public class RingBuffer {
    private static final Logger LOGGER = LoggerFactory.getLogger(RingBuffer.class);
    /** Constants */
    private static final int START_POINT = -1; 
    private static final long CAN_PUT_FLAG = 0L; //用于标记当前slot的状态，表示可以put一个id进去
    private static final long CAN_TAKE_FLAG = 1L; //用于标记当前slot的状态，表示可以take一个id
    public static final int DEFAULT_PADDING_PERCENT = 50; //用于控制何时填充slots的默认阈值：当剩余的可用的slot的个数，小于bufferSize的50%时，需要生成id将slots填满
    /** The size of RingBuffer's slots, each slot hold a UID */
    private final int bufferSize; //slots的大小，默认为sequence可容量的最大值，即8192个
    private final long indexMask; 
    private final long[] slots;  //slots用于缓存已经生成的id
    private final PaddedAtomicLong[] flags; //flags用于存储id的状态(是否可填充、是否可消费)
    /** Tail: last position sequence to produce */
    //Tail指针
    //表示Producer生产的最大序号(此序号从0开始，持续递增)。Tail不能超过Cursor，即生产者不能覆盖未消费的slot。当Tail已赶上curosr，此时可通过rejectedPutBufferHandler指定PutRejectPolicy
    private final AtomicLong tail = new PaddedAtomicLong(START_POINT); //
    /** Cursor: current position sequence to consume */
    //表示Consumer消费到的最小序号(序号序列与Producer序列相同)。Cursor不能超过Tail，即不能消费未生产的slot。当Cursor已赶上tail，此时可通过rejectedTakeBufferHandler指定TakeRejectPolicy
    private final AtomicLong cursor = new PaddedAtomicLong(START_POINT);
    /** Threshold for trigger padding buffer*/
    private final int paddingThreshold; //用于控制何时填充slots的阈值
    /** Reject put/take buffer handle policy */
    //当slots满了，无法继续put时的处理策略。默认实现：无法进行put，仅记录日志
    private RejectedPutBufferHandler rejectedPutHandler = this::discardPutBuffer;
    //当slots空了，无法继续take时的处理策略。默认实现：仅抛出异常
    private RejectedTakeBufferHandler rejectedTakeHandler = this::exceptionRejectedTakeBuffer; 
    /** Executor of padding buffer */
    //用于运行【生成id将slots填满】任务
    private BufferPaddingExecutor bufferPaddingExecutor;
复制代码

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;
    }
复制代码

// 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;
}
复制代码

//数组在物理上是连续存储的，flags数组用来保存id的状态（是否可消费、是否可填充），在填入id和消费id时，会被频繁的修改。
    //如果不进行缓存行填充，会导致频繁的缓存行失效，直接从内存中读数据。
    private final PaddedAtomicLong[] flags;
    //tail和cursor都使用缓存行填充，是为了避免tail和cursor落到同一个缓存行上。
    /** Tail: last position sequence to produce */
    private final AtomicLong tail = new PaddedAtomicLong(START_POINT);
    /** Cursor: current position sequence to consume */
    private final AtomicLong cursor = new PaddedAtomicLong(START_POINT)
复制代码

/**
 * Represents a padded {@link AtomicLong} to prevent the FalseSharing problem<p>
 * 
 * The CPU cache line commonly be 64 bytes, here is a sample of cache line after padding:<br>
 * 64 bytes = 8 bytes (object reference) + 6 * 8 bytes (padded long) + 8 bytes (a long value)
 * @author yutianbao
 */
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;
    }
}
复制代码

DROP TABLE IF EXISTS WORKER_NODE;
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;
复制代码

<dependencies>
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
  </dependency>
  <dependency>
    <groupId>org.mybatis.spring.boot</groupId>
    <artifactId>mybatis-spring-boot-starter</artifactId>
    <version>2.1.0</version>
  </dependency>
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
  </dependency>
  <!--for Mysql-->
  <dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <scope>runtime</scope>
    <version>8.0.12</version>
  </dependency>
  <!-- druid -->
  <dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>druid-spring-boot-starter</artifactId>
    <version>1.1.9</version>
  </dependency>
  <!--必须放在最后-->
  <dependency>
    <groupId>com.baidu.fsg</groupId>
    <artifactId>uid-generator</artifactId>
    <version>1.0.0-SNAPSHOT</version>
  </dependency>
</dependencies>
复制代码

<dependency>
    <groupId>com.xfvape.uid</groupId>
    <artifactId>uid-generator</artifactId>
    <version>0.0.4-RELEASE</version>
</dependency>
复制代码

<dependency>
    <groupId>com.baidu.fsg</groupId>
    <artifactId>uid-generator</artifactId>
    <version>1.0.0-SNAPSHOT</version>
    <exclusions>
        <exclusion>
            <groupId>org.mybatis</groupId>
            <artifactId>*</artifactId>
        </exclusion>
    </exclusions>
</dependency>
复制代码

<dependency>
    <groupId>com.xfvape.uid</groupId>
    <artifactId>uid-generator</artifactId>
    <version>0.0.4-RELEASE</version>
    <exclusions>
        <exclusion>
            <groupId>org.mybatis</groupId>
            <artifactId>*</artifactId>
        </exclusion>
    </exclusions>
</dependency>
复制代码

server.port=9999
spring.datasource.url=jdbc:mysql://*.*.*.*:3306/baiduUidGenerator?useUnicode=true&characterEncoding=utf-8&useSSL=false
spring.datasource.username=root
spring.datasource.password=*
spring.datasource.driver-class-name=com.mysql.jdbc.Driver
mybatis.mapper-locations=classpath:mapper/*.xml
mybatis.configuration.map-underscore-to-camel-case=true
复制代码

@MapperScan({"com.xxx.xx.dao"})
复制代码

/**
 * Represents an implementation of {@link WorkerIdAssigner},
 * the worker id will be discarded after assigned to the UidGenerator
 * 
 * @author yutianbao
 */
public class DisposableWorkerIdAssigner implements WorkerIdAssigner {
    private static final Logger LOGGER = LoggerFactory.getLogger(DisposableWorkerIdAssigner.class);
    @Autowired
    private WorkerNodeMapper workerNodeMapper;
    /**
     * Assign worker id base on database.<p>
     * If there is host name & port in the environment, we considered that the node runs in Docker container<br>
     * Otherwise, the node runs on an actual machine.
     * 
     * @return assigned worker id
     */
    @Override
    @Transactional
    public long assignWorkerId() {
        // build worker node entity
        WorkerNodeEntity workerNodeEntity = buildWorkerNode();
        // add worker node for new (ignore the same IP + PORT)
        workerNodeMapper.addWorkerNode(workerNodeEntity);
        LOGGER.info("Add worker node:" + workerNodeEntity);
        return workerNodeEntity.getId();
    }
    /**
     * Build worker node entity by IP and PORT
     */
    private WorkerNodeEntity buildWorkerNode() {
        WorkerNodeEntity workerNodeEntity = new WorkerNodeEntity();
        if (DockerUtils.isDocker()) {
            workerNodeEntity.setType(WorkerNodeType.CONTAINER.value());
            workerNodeEntity.setHostName(DockerUtils.getDockerHost());
            workerNodeEntity.setPort(DockerUtils.getDockerPort());
        } else {
            workerNodeEntity.setType(WorkerNodeType.ACTUAL.value());
            workerNodeEntity.setHostName(NetUtils.getLocalAddress());
            workerNodeEntity.setPort(System.currentTimeMillis() + "-" + RandomUtils.nextInt(100000));
        }
        return workerNodeEntity;
    }
}
复制代码

/**
 * <p> 百度  Uid-Generator配置
 * @author zepal
 * */
@Configuration
public class UidGeneratorConfig {
@Bean("disposableWorkerIdAssigner")
public DisposableWorkerIdAssigner disposableWorkerIdAssigner(){
  DisposableWorkerIdAssigner disposableWorkerIdAssigner = new DisposableWorkerIdAssigner();
  return  disposableWorkerIdAssigner;
}
@Bean("cachedUidGenerator")
  public CachedUidGenerator initCachedUidGenerator(WorkerIdAssigner workerIdAssigner) {
    CachedUidGenerator cachedUidGenerator = new CachedUidGenerator();
    cachedUidGenerator.setWorkerIdAssigner(workerIdAssigner);
    // 属性参考链接  https://github.com/baidu/uid-generator/blob/master/README.zh_cn.md
        // 以下为可选配置, 如未指定将采用默认值
    // cachedUidGenerator.setTimeBits(28);
    // cachedUidGenerator.setWorkerBits(22);
    // cachedUidGenerator.setSeqBits(13);
    // cachedUidGenerator.setEpochStr("2016-09-20");
    cachedUidGenerator.setBoostPower(3);
    cachedUidGenerator.setPaddingFactor(50);
    cachedUidGenerator.setScheduleInterval(60L);
//    // 拒绝策略: 当环已满, 无法继续填充时
    // 默认无需指定, 将丢弃Put操作, 仅日志记录. 如有特殊需求, 请实现RejectedPutBufferHandler接口(支持Lambda表达式)
    // 拒绝策略: 当环已空, 无法继续获取时
    // 默认无需指定, 将记录日志, 并抛出UidGenerateException异常. 如有特殊需求, 请实现RejectedTakeBufferHandler接口(支持Lambda表达式)
    return cachedUidGenerator;
  }
}
复制代码

/**
     * disposableWorkerIdAssigner的入参对象类型最好使用 WorkerIdAssigner，
     * 否则其他地方引入CGLib动态代理的时候可能会导致代理混用的问题
     *
     * @param disposableWorkerIdAssigner
     * @return
     */
    @Bean
    public DefaultUidGenerator defaultUidGenerator(WorkerIdAssigner disposableWorkerIdAssigner) {
        DefaultUidGenerator defaultUidGenerator = new DefaultUidGenerator();
     defaultUidGenerator.setWorkerIdAssigner(disposableWorkerIdAssigner);
        /**
         * 关于UID比特分配的建议：
         * 对于并发数要求不高、期望长期使用的应用, 可增加timeBits位数, 减少seqBits位数.
         * 例如节点采取用完即弃的WorkerIdAssigner策略, 重启频率为12次/天, 那么配置成：
         * {"workerBits":23,"timeBits":31,"seqBits":9}时, 可支持28个节点以整体并发量14400 UID/s的速度持续运行68年.
         *
         * 对于节点重启频率频繁、期望长期使用的应用, 可增加workerBits和timeBits位数, 减少seqBits位数.
         * 例如节点采取用完即弃的WorkerIdAssigner策略, 重启频率为24*12次/天, 那么配置成：
         * {"workerBits":27,"timeBits":30,"seqBits":6}时, 可支持37个节点以整体并发量2400 UID/s的速度持续运行34年.
         */
        //以下为可选配置, 如未指定将采用默认值
        defaultUidGenerator.setTimeBits(32);
        // 机器id，最多可支持2^22约420w次机器启动。内置实现为在启动时由数据库分配，默认分配策略为用后即弃，后续可提供复用策略。
        defaultUidGenerator.setWorkerBits(22);
        // 每秒下的并发序列，9 bits可支持每台服务器每秒512个并发。
        defaultUidGenerator.setSeqBits(9);
        defaultUidGenerator.setEpochStr("2020-01-01");
        return defaultUidGenerator;
    }
  /**
     * disposableWorkerIdAssigner的入参对象类型最好使用 WorkerIdAssigner，
     * 否则其他地方引入CGLib动态代理的时候可能会导致代理混用的问题
     *
     * @param disposableWorkerIdAssigner
     * @return
     */
    @Bean
    public CachedUidGenerator cachedUidGenerator(WorkerIdAssigner disposableWorkerIdAssigner) {
        CachedUidGenerator cachedUidGenerator = new CachedUidGenerator();
        cachedUidGenerator.setWorkerIdAssigner(disposableWorkerIdAssigner);
        /**
         * 关于UID比特分配的建议：
         * 对于并发数要求不高、期望长期使用的应用, 可增加timeBits位数, 减少seqBits位数.
         * 例如节点采取用完即弃的WorkerIdAssigner策略, 重启频率为12次/天, 那么配置成：
         * {"workerBits":23,"timeBits":31,"seqBits":9}时, 可支持28个节点以整体并发量14400 UID/s的速度持续运行68年.
         *
         * 对于节点重启频率频繁、期望长期使用的应用, 可增加workerBits和timeBits位数, 减少seqBits位数.
         * 例如节点采取用完即弃的WorkerIdAssigner策略, 重启频率为24*12次/天, 那么配置成：
         * {"workerBits":27,"timeBits":30,"seqBits":6}时, 可支持37个节点以整体并发量2400 UID/s的速度持续运行34年.
         */
        //以下为可选配置, 如未指定将采用默认值
        cachedUidGenerator.setTimeBits(32);
        // 机器id，最多可支持2^22约420w次机器启动。内置实现为在启动时由数据库分配，默认分配策略为用后即弃，后续可提供复用策略。
        cachedUidGenerator.setWorkerBits(22);
        // 每秒下的并发序列，9 bits可支持每台服务器每秒512个并发。
        cachedUidGenerator.setSeqBits(9);
        cachedUidGenerator.setEpochStr("2020-01-01");
        //RingBuffer size扩容参数, 可提高UID生成的吞吐量
        //默认:3， 原bufferSize=8192, 扩容后bufferSize= 8192 << 3 = 65536
        cachedUidGenerator.setBoostPower(3);
        // 指定何时向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线程时间间隔, 单位:秒
        cachedUidGenerator.setScheduleInterval(60L);
        //拒绝策略: 当环已满, 无法继续填充时
        //默认无需指定, 将丢弃Put操作, 仅日志记录. 如有特殊需求, 请实现RejectedPutBufferHandler接口(支持Lambda表达式)
        //<property name="rejectedPutBufferHandler" ref="XxxxYourPutRejectPolicy"></property>
        //cachedUidGenerator.setRejectedPutBufferHandler();
        //拒绝策略: 当环已空, 无法继续获取时 -->
        //默认无需指定, 将记录日志, 并抛出UidGenerateException异常. 如有特殊需求, 请实现RejectedTakeBufferHandler接口(支持Lambda表达式) -->
        //<property name="rejectedTakeBufferHandler" ref="XxxxYourTakeRejectPolicy"></property>
        return cachedUidGenerator;
    }
复制代码

@Repository
public interface WorkerNodeMapper {
    /**
     * Get {@link WorkerNodeEntity} by node host
     * 
     * @param host
     * @param port
     * @return
     */
    WorkerNodeEntity getWorkerNodeByHostPort(@Param("host") String host, @Param("port") String port);
    /**
     * Add {@link WorkerNodeEntity}
     * 
     * @param workerNodeEntity
     */
    void addWorkerNode(WorkerNodeEntity workerNodeEntity);
}
复制代码

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="org.zxp.uidgeneratortest.WorkerNodeMapper">
    <resultMap id="workerNodeRes"
               type="com.baidu.fsg.uid.worker.entity.WorkerNodeEntity">
        <id column="ID" jdbcType="BIGINT" property="id"/>
        <result column="HOST_NAME" jdbcType="VARCHAR" property="hostName"/>
        <result column="PORT" jdbcType="VARCHAR" property="port"/>
        <result column="TYPE" jdbcType="INTEGER" property="type"/>
        <result column="LAUNCH_DATE" jdbcType="DATE" property="launchDate"/>
        <result column="MODIFIED" jdbcType="TIMESTAMP" property="modified"/>
        <result column="CREATED" jdbcType="TIMESTAMP" property="created"/>
    </resultMap>
    <insert id="addWorkerNode" useGeneratedKeys="true" keyProperty="id"
            parameterType="com.baidu.fsg.uid.worker.entity.WorkerNodeEntity">
    INSERT INTO WORKER_NODE
    (HOST_NAME,
    PORT,
    TYPE,
    LAUNCH_DATE,
    MODIFIED,
    CREATED)
    VALUES (
    #{hostName},
    #{port},
    #{type},
    #{launchDate},
    NOW(),
    NOW())
  </insert>
    <select id="getWorkerNodeByHostPort" resultMap="workerNodeRes">
    SELECT
    ID,
    HOST_NAME,
    PORT,
    TYPE,
    LAUNCH_DATE,
    MODIFIED,
    CREATED
    FROM
    WORKER_NODE
    WHERE
    HOST_NAME = #{host} AND PORT = #{port}
  </select>
</mapper>
复制代码

@Service
public class UidGenService {
    @Resource
    private UidGenerator uidGenerator;
    public long getUid() {
        return uidGenerator.getUID();
    }
}