python雪花算法

雪花算法（Snowflake Algorithm）是一种用于生成唯一的ID的算法，它由Twitter开发。其生成的ID在全局范围内是唯一的，适合高并发场景。雪花算法生成的ID通常是一个64位的整数，包含多个部分，具体结构如下：

1. **时间戳（41位）**：当前时间的毫秒数，能支持69年的时间范围。
2. **工作机器ID（10位）**：机器ID，可以支持1024台机器。
3. **数据中心ID（10位）**：数据中心ID，支持1024个数据中心。
4. **序列号（12位）**：在同一毫秒内生成的ID序列号，支持每毫秒生成4096个ID。

以下是一个用Python实现的雪花算法的代码示例，附带注释：

import time
import threading

class Snowflake:
# 基本参数
epoch = 1609459200000 # 自定义的开始时间（2021-01-01 00:00:00）
worker_id_bits = 10 # 工作机器ID占用的位数
datacenter_id_bits = 10 # 数据中心ID占用的位数
sequence_bits = 12 # 序列号占用的位数

# 计算各部分的位移
worker_id_shift = sequence_bits # 序列号位移
datacenter_id_shift = sequence_bits + worker_id_bits # 数据中心ID位移
timestamp_shift = sequence_bits + worker_id_bits + datacenter_id_bits # 时间戳位移

# 生成掩码
max_worker_id = -1 ^ (-1 << worker_id_bits) # 计算最大工作机器ID
max_datacenter_id = -1 ^ (-1 << datacenter_id_bits) # 计算最大数据中心ID
sequence_mask = -1 ^ (-1 << sequence_bits) # 计算序列号掩码

def __init__(self, worker_id, datacenter_id):
if worker_id > self.max_worker_id or worker_id < 0:
raise ValueError(f'worker_id必须在0到{self.max_worker_id}之间')
if datacenter_id > self.max_datacenter_id or datacenter_id < 0:
raise ValueError(f'datacenter_id必须在0到{self.max_datacenter_id}之间')

self.worker_id = worker_id
self.datacenter_id = datacenter_id
self.sequence = 0 # 序列号初始值
self.last_timestamp = -1 # 上一次生成ID的时间戳

def _current_time_millis(self):
return int(time.time() * 1000) # 获取当前时间戳（毫秒）

def generate_id(self):
timestamp = self._current_time_millis() # 获取当前时间戳

if timestamp < self.last_timestamp:
raise Exception('时钟向后移动，无法生成ID')

if self.last_timestamp == timestamp:
self.sequence = (self.sequence + 1) & self.sequence_mask # 在同一毫秒内序列号加1
if self.sequence == 0:
# 序列号已满，等待下一毫秒
while timestamp <= self.last_timestamp:
timestamp = self._current_time_millis()
else:
self.sequence = 0 # 重置序列号

self.last_timestamp = timestamp # 更新上一次时间戳

# 生成ID
id = ((timestamp - self.epoch) << self.timestamp_shift) | \
(self.datacenter_id << self.datacenter_id_shift) | \
(self.worker_id << self.worker_id_shift) | \
self.sequence
return id

# 示例：创建一个雪花算法实例并生成ID
if __name__ == "__main__":
worker_id = 1 # 机器ID
datacenter_id = 1 # 数据中心ID
snowflake = Snowflake(worker_id, datacenter_id)

# 生成10个ID
for _ in range(10):
print(snowflake.generate_id())

### 代码说明：
– **Snowflake 类**：主要实现雪花算法的逻辑。
– **__init__ 方法**：初始化机器ID和数据中心ID，检查其合法性。
– **_current_time_millis 方法**：获取当前时间的毫秒数。
– **generate_id 方法**：生成唯一的ID，处理序列号的自增和时间戳的更新。
– **主程序**：创建一个 `Snowflake` 实例并生成10个唯一的ID。

这种算法适合分布式系统中生成唯一标识符，确保在高并发的情况下仍然能生成不重复的ID。

更多实用教程资源：http://sj.ysok.net/jydoraemon 访问码：JYAM

来源链接：https://www.cnblogs.com/HQING/p/18603618