标签：this eid 细节 cal handler red 全局 个数 each
一提到分布式ID自动生成方案，大家肯定都非常熟悉，并且立即能说出自家拿手的几种方案，确实，ID作为系统数据的重要标识，重要性不言而喻，而各种方案也是历经多代优化，请允许我用这个视角对分布式ID自动生成方案进行分类：
ID的生成规则，读取控制完全由数据源控制，常见的如数据库的自增长ID，序列号等，或Redis的INCR/INCRBY原子操作产生顺序号等。
ID的生成规则，有部分生成因子需要由数据源（或配置信息）控制，如snowflake算法。
ID的生成规则完全由机器信息独立计算，不依赖任何配置信息和数据记录，如常见的UUID，GUID等
实践方案适用于以上提及的三种实现方式，可作为这三种实现方式的一种补充，旨在提升系统吞吐量，但原有实现方式的局限性依然存在。
顾名思义，每次要获取ID时，实时生成。
简单快捷，ID都是连续不间断的，但吞吐量可能不是最高。
预先生成一批ID放在数据池里，可简单自增长生成，也可以设置步长，分批生成，需要将这些预先生成的数据，放在存储容器里（JVM内存，Redis，数据库表均可）。
可以较大幅度地提升吞吐量，但需要开辟临时存储空间，断电宕机后可能会丢失已有ID，ID可能有间断。
以下对目前流行的分布式ID方案做简单介绍
属于完全依赖数据源的方式，所有的ID存储在数据库里，是最常用的ID生成办法，在单体应用时期得到了最广泛的使用，建立数据表时利用数据库自带的auto_increment作主键，或是使用序列完成其他场景的一些自增长ID的需求。
也属于完全依赖数据源的方式，通过Redis的INCR/INCRBY自增原子操作命令，能保证生成的ID肯定是唯一有序的，本质上实现方式与数据库一致。
UUID：按照OSF制定的标准计算，用到了以太网卡地址、纳秒级时间、芯片ID码和许多可能的数字。由以下几部分的组合：当前日期和时间(UUID的第一个部分与时间有关，如果你在生成一个UUID之后，过几秒又生成一个UUID，则第一个部分不同，其余相同)，时钟序列，全局唯一的IEEE机器识别号（如果有网卡，从网卡获得，没有网卡以其他方式获得）
GUID：微软对UUID这个标准的实现。UUID还有其它各种实现，不止GUID一种，不一一列举了。
这两种属于不依赖数据源方式，真正的全球唯一性ID
4、snowflake算法（雪花算法）生成ID
属于半依赖数据源方式，原理是使用Long类型（64位），按照一定的规则进行填充：时间（毫秒级）+集群ID+机器ID+序列号，每部分占用的位数可以根据实际需要分配，其中集群ID和机器ID这两部分，在实际应用场景中要依赖外部参数配置或数据库记录。
雪花ID算法听起来是不是特别适用分布式架构场景？照目前来看是的，接下来我们重点讲解它的原理和最佳实践。
snowflake算法来源于Twitter，使用scala语言实现，利用Thrift框架实现RPC接口调用，最初的项目起因是数据库从mysql迁移到Cassandra，Cassandra没有现成可用 的ID生成机制，就催生了这个项目，现有的github源码有兴趣可以去看看。
snowflake算法的特性是有序、唯一，并且要求高性能，低延迟（每台机器每秒至少生成10k条数据，并且响应时间在2ms以内），要在分布式环境（多集群，跨机房）下使用，因此snowflake算法得到的ID是分段组成的：
如图所示：
以上的位数分配只是官方建议的，我们可以根据实际需要自行分配，比如说我们的应用机器数量最多也就几十台，但并发数很大，我们就可以将10bit减少到8bit，序列部分从12bit增加到14bit等等
当然每部分的含义也可以自由替换，如中间部分的机器ID，如果是云计算、容器化的部署环境，随时有扩容，缩减机器的操作，通过线下规划去配置实例的ID不太现实，就可以替换为实例每重启一次，拿一次自增长的ID作为该部分的内容，下文会讲解。
github上也有大神用Java做了snowflake最基本的实现，这里直接查看源码：
snowflake java版源码
/**
* twitter的snowflake算法 -- java实现
*
* @author beyond
* @date 2016/11/26
*/
public class SnowFlake {
/**
* 起始的时间戳
*/
private final static long START_STMP = 1480166465631L;
/**
* 每一部分占用的位数
*/
private final static long SEQUENCE_BIT = 12; //序列号占用的位数
private final static long MACHINE_BIT = 5; //机器标识占用的位数
private final static long DATACENTER_BIT = 5;//数据中心占用的位数
/**
* 每一部分的最大值
*/
private final static long MAX_DATACENTER_NUM = -1L ^ (-1L << DATACENTER_BIT);
private final static long MAX_MACHINE_NUM = -1L ^ (-1L << MACHINE_BIT);
private final static long MAX_SEQUENCE = -1L ^ (-1L << SEQUENCE_BIT);
/**
* 每一部分向左的位移
*/
private final static long MACHINE_LEFT = SEQUENCE_BIT;
private final static long DATACENTER_LEFT = SEQUENCE_BIT + MACHINE_BIT;
private final static long TIMESTMP_LEFT = DATACENTER_LEFT + DATACENTER_BIT;
private long datacenterId; //数据中心
private long machineId; //机器标识
private long sequence = 0L; //序列号
private long lastStmp = -1L;//上一次时间戳
public SnowFlake(long datacenterId, long machineId) {
if (datacenterId > MAX_DATACENTER_NUM || datacenterId < 0) {
throw new IllegalArgumentException("datacenterId can't be greater than MAX_DATACENTER_NUM or less than 0");
}
if (machineId > MAX_MACHINE_NUM || machineId < 0) {
throw new IllegalArgumentException("machineId can't be greater than MAX_MACHINE_NUM or less than 0");
}
this.datacenterId = datacenterId;
this.machineId = machineId;
}
/**
* 产生下一个ID
*
* @return
*/
public synchronized long nextId() {
long currStmp = getNewstmp();
if (currStmp < lastStmp) {
throw new RuntimeException("Clock moved backwards. Refusing to generate id");
}
if (currStmp == lastStmp) {
//相同毫秒内，序列号自增
sequence = (sequence + 1) & MAX_SEQUENCE;
//同一毫秒的序列数已经达到最大
if (sequence == 0L) {
currStmp = getNextMill();
}
} else {
//不同毫秒内，序列号置为0
sequence = 0L;
}
lastStmp = currStmp;
return (currStmp - START_STMP) << TIMESTMP_LEFT //时间戳部分
| datacenterId << DATACENTER_LEFT //数据中心部分
| machineId << MACHINE_LEFT //机器标识部分
| sequence; //序列号部分
}
private long getNextMill() {
long mill = getNewstmp();
while (mill <= lastStmp) {
mill = getNewstmp();
}
return mill;
}
private long getNewstmp() {
return System.currentTimeMillis();
}
public static void main(String[] args) {
SnowFlake snowFlake = new SnowFlake(2, 3);
for (int i = 0; i < (1 << 12); i++) {
System.out.println(snowFlake.nextId());
}
}
}基本上通过位移操作，将每段含义的数值，移到相应的位置上，如机器ID这里由数据中心+机器标识组成，所以，机器标识向左移12位，就是它的位置，数据中心的编号向左移17位，时间戳的值向左移22位，每部分占据自己的位置，各不干涉，由此组成一个完整的ID值。
这里就是snowflake最基础的实现原理，如果有些java基础知识不记得了建议查一下资料，如二进制-1的表示是0xffff(里面全是1)，<<表示左移操作，-1<<5等于-32，异或操作-1 ^ (-1 << 5)为31等等。
了解snowflake的基本实现原理，可以通过提前规划好机器标识来实现，但目前的分布式生产环境，借用了多种云计算、容器化技术，实例的个数随时有变化，还需要处理服务器实例时钟回拨的问题，固定规划ID然后通过配置来使用snowflake的场景可行性不高，一般是自动启停，增减机器，这样就需要对snowflake进行一些改造才能更好地应用到生产环境中。
UidGenerator项目基于snowflake原理实现，只是修改了机器ID部分的定义（实例重启的次数），并且64位bit的分配支持配置，官方提供的默认分配方式如下图：
Snowflake算法描述：指定机器 & 同一时刻 & 某一并发序列，是唯一的。据此可生成一个64 bits的唯一ID（long）。
具体的实现有两种，一种是实时生成ID，另一种是预先生成ID方式
核心代码如下：
/**
* 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);
}机器ID的获取方法与上一种相同，这种是预先生成一批ID，放在一个RingBuffer环形数组里，供客户端使用，当可用数据低于阀值时，再次调用批量生成方法，属于用空间换时间的做法，可以提高整个ID的吞吐量。
核心代码：
/**
* Initialize RingBuffer & RingBufferPaddingExecutor
*/
private void initRingBuffer() {
// initialize RingBuffer
int bufferSize = ((int) bitsAllocator.getMaxSequence() + 1) << boostPower;
this.ringBuffer = new RingBuffer(bufferSize, paddingFactor);
LOGGER.info("Initialized ring buffer size:{}, paddingFactor:{}", bufferSize, paddingFactor);
// initialize RingBufferPaddingExecutor
boolean usingSchedule = (scheduleInterval != null);
this.bufferPaddingExecutor = new BufferPaddingExecutor(ringBuffer, this::nextIdsForOneSecond, usingSchedule);
if (usingSchedule) {
bufferPaddingExecutor.setScheduleInterval(scheduleInterval);
}
LOGGER.info("Initialized BufferPaddingExecutor. Using schdule:{}, interval:{}", usingSchedule, scheduleInterval);
// set rejected put/take handle policy
this.ringBuffer.setBufferPaddingExecutor(bufferPaddingExecutor);
if (rejectedPutBufferHandler != null) {
this.ringBuffer.setRejectedPutHandler(rejectedPutBufferHandler);
}
if (rejectedTakeBufferHandler != null) {
this.ringBuffer.setRejectedTakeHandler(rejectedTakeBufferHandler);
}
// fill in all slots of the RingBuffer
bufferPaddingExecutor.paddingBuffer();
// start buffer padding threads
bufferPaddingExecutor.start();
}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;
}
核心代码：
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;
}另外有个细节可以了解一下，RingBuffer的数据都是使用数组来存储的，考虑CPU Cache的结构，tail和cursor变量如果直接用原生的AtomicLong类型，tail和cursor可能会缓存在同一个cacheLine中，多个线程读取该变量可能会引发CacheLine的RFO请求，反而影响性能,为了防止伪共享问题，特意填充了6个long类型的成员变量，加上long类型的value成员变量，刚好占满一个Cache Line（Java对象还有8byte的对象头），这个叫CacheLine补齐，有兴趣可以了解一下，源码如下：
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;
}
}以上是百度uid-generator项目的主要描述，我们可以发现，snowflake算法在落地时有一些变化，主要体现在机器ID的获取上，尤其是分布式集群环境下面，实例自动伸缩，docker容器化的一些技术，使得静态配置项目ID，实例ID可行性不高，所以这些转换为按启动次数来标识。
在uidGenerator方面，美团的项目源码直接集成百度的源码，略微将一些Lambda表达式换成原生的java语法，例如：
// com.myzmds.ecp.core.uid.baidu.impl.CachedUidGenerator类的initRingBuffer()方法
// 百度源码
this.bufferPaddingExecutor = new BufferPaddingExecutor(ringBuffer, this::nextIdsForOneSecond, usingSchedule);
// 美团源码
this.bufferPaddingExecutor = new BufferPaddingExecutor(ringBuffer, new BufferedUidProvider() {
@Override
public List<Long> provide(long momentInSecond) {
return nextIdsForOneSecond(momentInSecond);
}
}, usingSchedule);并且在机器ID生成方面，引入了Zookeeper，Redis这些组件，丰富了机器ID的生成和获取方式，实例编号可以存储起来反复使用，不再是数据库单调增长这一种了。
本篇简单介绍了snowflake算法的原理及落地过程中的改造，在此学习了优秀的开源代码，并挑出部分进行了简单的示例，美团的ecp-uid项目不但集成了百度现有的UidGenerator算法，原生的snowflake算法，还包含优秀的leaf segment算法，鉴于篇幅没有详尽描述。文章内有任何不正确或不详尽之处请留言指出，谢谢。
专注Java高并发、分布式架构，更多技术干货分享与心得，请关注公众号：Java架构社区
标签：this eid 细节 cal handler red 全局 个数 each
原文地址：https://www.cnblogs.com/huangying2124/p/11736031.html
