牵着老婆满街逛

严以律己,宽以待人. 三思而后行.
GMail/GTalk: yanglinbo#google.com;
MSN/Email: tx7do#yahoo.com.cn;
QQ: 3 0 3 3 9 6 9 2 0 .

Twitter的分布式自增ID算法Snowflake实现分析及其Java、Php和Python版

转载自:http://www.dengchuanhua.com/132.html

在分布式系统中,需要生成全局UID的场合还是比较多的,twitter的snowflake解决了这种需求,实现也还是很简单的,除去配置信息,核心代码就是毫秒级时间41位+机器ID 10位+毫秒内序列12位。

该项目地址为:https://github.com/twitter/snowflake是用Scala实现的。

python版详见开源项目https://github.com/erans/pysnowflake

核心代码为其IdWorker这个类实现,其原理结构如下,我分别用一个0表示一位,用—分割开部分的作用:

1 0---0000000000 0000000000 0000000000 0000000000 0 --- 00000 ---00000 ---000000000000

在上面的字符串中,第一位为未使用(实际上也可作为long的符号位),接下来的41位为毫秒级时间,然后5位datacenter标识位,5位机器ID(并不算标识符,实际是为线程标识),然后12位该毫秒内的当前毫秒内的计数,加起来刚好64位,为一个Long型。

这样的好处是,整体上按照时间自增排序,并且整个分布式系统内不会产生ID碰撞(由datacenter和机器ID作区分),并且效率较高,经测试,snowflake每秒能够产生26万ID左右,完全满足需要。

且看其核心代码:

1 </pre>
2 /** Copyright 2010-2012 Twitter, Inc.*/
3 package com.twitter.service.snowflake
4  
5 import com.twitter.ostrich.stats.Stats
6 import com.twitter.service.snowflake.gen._
7 import java.util.Random
8 import com.twitter.logging.Logger
9  
10 /**
11  * An object that generates IDs.
12  * This is broken into a separate class in case
13  * we ever want to support multiple worker threads
14  * per process
15  */
16 class IdWorker(val workerId: Long, val datacenterId: Long, private val reporter: Reporter, var sequence: Long = 0L)
17 extends Snowflake.Iface {
18   private[this] def genCounter(agent: String) = {
19     Stats.incr("ids_generated")
20     Stats.incr("ids_generated_%s".format(agent))
21   }
22   private[this] val exceptionCounter = Stats.getCounter("exceptions")
23   private[this] val log = Logger.get
24   private[this] val rand = new Random
25  
26   val twepoch = 1288834974657L
27  
28  //机器标识位数
29  
30   private[this] val workerIdBits = 5L
31  
32 //数据中心标识位数
33   private[this] val datacenterIdBits = 5L
34  
35 //机器ID最大值
36   private[this] val maxWorkerId = -1L ^ (-1L << workerIdBits)
37  
38 //数据中心ID最大值
39   private[this] val maxDatacenterId = -1L ^ (-1L << datacenterIdBits)
40  
41 //毫秒内自增位
42   private[this] val sequenceBits = 12L
43  
44 //机器ID偏左移12位
45  
46   private[this] val workerIdShift = sequenceBits
47  
48 //数据中心ID左移17位
49   private[this] val datacenterIdShift = sequenceBits + workerIdBits
50  
51 //时间毫秒左移22位
52   private[this] val timestampLeftShift = sequenceBits + workerIdBits + datacenterIdBits
53   private[this] val sequenceMask = -1L ^ (-1L << sequenceBits)
54  
55   private[this] var lastTimestamp = -1L
56  
57   // sanity check for workerId
58   if (workerId > maxWorkerId || workerId < 0) {
59     exceptionCounter.incr(1)
60     throw new IllegalArgumentException("worker Id can't be greater than %d or less than 0".format(maxWorkerId))
61   }
62  
63   if (datacenterId > maxDatacenterId || datacenterId < 0) {
64     exceptionCounter.incr(1)
65     throw new IllegalArgumentException("datacenter Id can't be greater than %d or less than 0".format(maxDatacenterId))
66   }
67  
68   log.info("worker starting. timestamp left shift %d, datacenter id bits %d, worker id bits %d, sequence bits %d, workerid %d",
69     timestampLeftShift, datacenterIdBits, workerIdBits, sequenceBits, workerId)
70  
71   def get_id(useragent: String): Long = {
72     if (!validUseragent(useragent)) {
73       exceptionCounter.incr(1)
74       throw new InvalidUserAgentError
75     }
76  
77     val id = nextId()
78     genCounter(useragent)
79  
80     reporter.report(new AuditLogEntry(id, useragent, rand.nextLong))
81     id
82   }
83  
84   def get_worker_id(): Long = workerId
85   def get_datacenter_id(): Long = datacenterId
86   def get_timestamp() = System.currentTimeMillis
87  
88   protected[snowflake] def nextId(): Long = synchronized {
89     var timestamp = timeGen()
90  
91  //时间错误
92  
93     if (timestamp < lastTimestamp) {
94       exceptionCounter.incr(1)
95       log.error("clock is moving backwards.  Rejecting requests until %d.", lastTimestamp);
96       throw new InvalidSystemClock("Clock moved backwards.  Refusing to generate id for %d milliseconds".format(
97         lastTimestamp - timestamp))
98     }
99  
100     if (lastTimestamp == timestamp) {
101 //当前毫秒内,则+1
102       sequence = (sequence + 1) & sequenceMask
103       if (sequence == 0) {
104 //当前毫秒内计数满了,则等待下一秒
105         timestamp = tilNextMillis(lastTimestamp)
106       }
107     } else {
108       sequence = 0
109     }
110  
111     lastTimestamp = timestamp