使用雪花算法（Snowflake Algorithm）在Python中生成唯一ID

使用雪花算法Snowflake Algorithm在Python中生成唯一ID

使用雪花算法（Snowflake Algorithm）在Python中生成唯一ID
- 雪花算法简介
- Python实现
- 代码解析
- 使用示例
- 优势
- 注意事项
- 适用场景
- 结论

使用雪花算法（Snowflake Algorithm）在Python中生成唯一ID

在分布式系统中生成唯一ID是一个常见的需求。Twitter的雪花算法（Snowflake Algorithm）是一种优秀的解决方案，它可以生成64位的唯一ID，包含时间戳、工作机器ID和序列号。今天，我们将探讨如何在Python中实现雪花算法。

雪花算法简介

雪花算法生成的ID结构如下：

1位符号位（始终为0）
41位时间戳（毫秒级）
5位数据中心ID
5位工作机器ID
12位序列号

这种结构允许在同一毫秒内生成4096个唯一ID，并支持多个数据中心和工作机器。

Python实现

import time
import threadingclass SnowflakeGenerator:def __init__(self, datacenter_id, worker_id):self.datacenter_id = datacenter_idself.worker_id = worker_idself.sequence = 0self.last_timestamp = -1# Bit lengths for different partsself.datacenter_id_bits = 5self.worker_id_bits = 5self.sequence_bits = 12# Maximum valuesself.max_datacenter_id = -1 ^ (-1 << self.datacenter_id_bits)self.max_worker_id = -1 ^ (-1 << self.worker_id_bits)self.max_sequence = -1 ^ (-1 << self.sequence_bits)# Shift amountsself.worker_id_shift = self.sequence_bitsself.datacenter_id_shift = self.sequence_bits + self.worker_id_bitsself.timestamp_shift = self.sequence_bits + self.worker_id_bits + self.datacenter_id_bitsself.lock = threading.Lock()def _current_milliseconds(self):return int(time.time() * 1000)def _til_next_millis(self, last_timestamp):timestamp = self._current_milliseconds()while timestamp <= last_timestamp:timestamp = self._current_milliseconds()return timestampdef generate_id(self):with self.lock:timestamp = self._current_milliseconds()if timestamp < self.last_timestamp:raise ValueError("Clock moved backwards. Refusing to generate id.")if timestamp == self.last_timestamp:self.sequence = (self.sequence + 1) & self.max_sequenceif self.sequence == 0:timestamp = self._til_next_millis(self.last_timestamp)else:self.sequence = 0self.last_timestamp = timestampreturn ((timestamp - 1288834974657) << self.timestamp_shift) | \(self.datacenter_id << self.datacenter_id_shift) | \(self.worker_id << self.worker_id_shift) | \self.sequencedef generate_unique_id(prefix: str, datacenter_id: int, worker_id: int) -> str:generator = SnowflakeGenerator(datacenter_id, worker_id)snowflake_id = generator.generate_id()return f"{prefix}{snowflake_id}"

代码解析

SnowflakeGenerator类：
- 初始化方法设置数据中心ID和工作机器ID，并定义各部分的位长度和最大值。
- _current_milliseconds()方法获取当前时间戳（毫秒级）。
- _til_next_millis()方法等待直到下一毫秒。
- generate_id()方法是核心，它生成唯一的雪花ID。
generate_id()方法：
- 使用锁确保线程安全。
- 获取当前时间戳。
- 处理时钟回拨问题（抛出异常）。
- 处理同一毫秒内的序列号。
- 组合各部分生成最终的ID。
generate_unique_id()函数：
- 创建SnowflakeGenerator实例。
- 生成雪花ID并添加前缀。

使用示例

# 创建一个生成器，指定数据中心ID和工作机器ID
datacenter_id = 1
worker_id = 1# 生成用户ID
user_id = generate_unique_id("USER_", datacenter_id, worker_id)
print(f"生成的用户ID: {user_id}")# 生成订单ID
order_id = generate_unique_id("ORDER_", datacenter_id, worker_id)
print(f"生成的订单ID: {order_id}")# 生成多个产品ID
for i in range(5):product_id = generate_unique_id("PROD_", datacenter_id, worker_id)print(f"产品 {i+1} ID: {product_id}")

输出可能如下：

生成的用户ID: USER_6791951648483729408
生成的订单ID: ORDER_6791951648483729409
产品 1 ID: PROD_6791951648483729410
产品 2 ID: PROD_6791951648483729411
产品 3 ID: PROD_6791951648483729412
产品 4 ID: PROD_6791951648483729413
产品 5 ID: PROD_6791951648483729414