雪花算法(Snowflake)是一种用于生成唯一ID的分布式算法,由Twitter开源。它生成的ID是一个64位的长整型数字,结构如下:
+----------------------+
| 41 |
| 0 |
+----------------------+ // 4位时间戳
| 0 | // 5位工作机器ID
+----------------------+ // 5位序列号
| 0 | // 12位数据中心ID
+----------------------+
雪花算法生成的ID具有全局唯一性,且趋势递增,适用于分布式系统中生成唯一ID的场景。下面是一个简单的雪花算法实现:
```python import time import random
class Snowflake: def init(self, datacenter_id, worker_id): self.datacenter_id = datacenter_id self.worker_id = worker_id self.sequence = 0 self.last_timestamp = -1
self.twepoch = 1288834974657 # 开始时间戳 (2020-01-01)
def _til_next_millis(self, last_timestamp):
timestamp = int(time.time() * 1000)
while timestamp <= last_timestamp:
timestamp = int(time.time() * 1000)
return timestamp
def next_id(self):
timestamp = int(time.time() * 1000)
if timestamp < self.last_timestamp:
raise Exception("Clock moved backwards. Refusing to generate id for %d milliseconds" % (self.last_timestamp - timestamp))
if timestamp == self.last_timestamp:
self.sequence = (self.sequence + 1) & 4095
if self.sequence == 0:
timestamp = self._til_next_millis(self.last_timestamp)
else:
self.sequence = 0
self.last_timestamp = timestamp
return ((timestamp - self.twepoch) << 22) | (self.datacenter_id << 17) | (self.worker_id << 12) | self.sequence
示例
datacenter_id = 1 worker_id = 1 snowflake = Snowflake(datacenter_id, worker_id) print(snowflake.next_id()) ```
这个实现中,datacenter_id
和 worker_id
分别表示数据中心ID和工作机器ID,需要根据实际情况分配不同的值。next_id()
方法用于生成唯一ID。