注意机器码参数配置# 虚拟环境py3_8_20_patent_env # 自定义雪花算法 import time import threading class SnowflakeGenerator: def __init__(self, machine_id: int): 初始化雪花ID生成器 :param machine_id: 机器ID (0~1023)需确保不同节点唯一 if machine_id 0 or machine_id 1024: raise ValueError(machine_id 必须在 0~1023 之间) self.machine_id machine_id # 10位机器ID self.sequence 0 # 12位序列号 self.last_timestamp -1 # 上次生成时间戳 self.lock threading.Lock() # 线程锁确保线程安全 # 自定义起始时间可调整为系统上线时间延长ID有效期 self.epoch 1609459200000 # 默认2021-01-01 00:00:00 def generate(self) - int: 生成唯一ID with self.lock: timestamp self._current_time() # 处理时间回拨 if timestamp self.last_timestamp: raise ValueError(f时间回拨当前时间 {timestamp} 上次时间 {self.last_timestamp}) # 同一毫秒内生成 if timestamp self.last_timestamp: self.sequence (self.sequence 1) 0xFFF # 12位序列号自增 if self.sequence 0: # 当前毫秒序列号用尽 timestamp self._wait_next_millis() else: self.sequence 0 self.last_timestamp timestamp # 组合各部分的二进制位 return ( ((timestamp - self.epoch) 22) | # 41位时间差 (self.machine_id 12) | # 10位机器ID self.sequence) # 12位序列号 def parse(self, snowflake_id: int) - dict: 解析雪花ID结构 :return: 包含时间戳、机器ID、序列号的字典 timestamp (snowflake_id 22) self.epoch machine_id (snowflake_id 12) 0x3FF # 0x3FF 二进制 1111111111 (10位) sequence snowflake_id 0xFFF # 0xFFF 二进制 111111111111 (12位) return { timestamp: timestamp, machine_id: machine_id, sequence: sequence, datetime: time.strftime(%Y-%m-%d %H:%M:%S, time.localtime(timestamp / 1000)) } def _current_time(self) - int: 获取当前时间戳毫秒 return int(time.time() * 1000) def _wait_next_millis(self) - int: 等待下一毫秒 timestamp self._current_time() while timestamp self.last_timestamp: time.sleep(0.001) timestamp self._current_time() return timestamp # 使用示例 if __name__ __main__: # 初始化生成器machine_id 需全局唯一 generator SnowflakeGenerator(machine_id1) # 生成ID snowflake_id generator.generate() print(f生成的ID: {snowflake_id}) # 输出如 1318958301703372800 # 解析ID print(解析结果:, generator.parse(snowflake_id)) # 输出: {timestamp: 1715586304000, machine_id: 1, sequence: 0, datetime: 2024-05-12 14:25:04}
注意机器码参数配置# 虚拟环境py3_8_20_patent_env # 自定义雪花算法 import time import threading class SnowflakeGenerator: def __init__(self, machine_id: int): 初始化雪花ID生成器 :param machine_id: 机器ID (0~1023)需确保不同节点唯一 if machine_id 0 or machine_id 1024: raise ValueError(machine_id 必须在 0~1023 之间) self.machine_id machine_id # 10位机器ID self.sequence 0 # 12位序列号 self.last_timestamp -1 # 上次生成时间戳 self.lock threading.Lock() # 线程锁确保线程安全 # 自定义起始时间可调整为系统上线时间延长ID有效期 self.epoch 1609459200000 # 默认2021-01-01 00:00:00 def generate(self) - int: 生成唯一ID with self.lock: timestamp self._current_time() # 处理时间回拨 if timestamp self.last_timestamp: raise ValueError(f时间回拨当前时间 {timestamp} 上次时间 {self.last_timestamp}) # 同一毫秒内生成 if timestamp self.last_timestamp: self.sequence (self.sequence 1) 0xFFF # 12位序列号自增 if self.sequence 0: # 当前毫秒序列号用尽 timestamp self._wait_next_millis() else: self.sequence 0 self.last_timestamp timestamp # 组合各部分的二进制位 return ( ((timestamp - self.epoch) 22) | # 41位时间差 (self.machine_id 12) | # 10位机器ID self.sequence) # 12位序列号 def parse(self, snowflake_id: int) - dict: 解析雪花ID结构 :return: 包含时间戳、机器ID、序列号的字典 timestamp (snowflake_id 22) self.epoch machine_id (snowflake_id 12) 0x3FF # 0x3FF 二进制 1111111111 (10位) sequence snowflake_id 0xFFF # 0xFFF 二进制 111111111111 (12位) return { timestamp: timestamp, machine_id: machine_id, sequence: sequence, datetime: time.strftime(%Y-%m-%d %H:%M:%S, time.localtime(timestamp / 1000)) } def _current_time(self) - int: 获取当前时间戳毫秒 return int(time.time() * 1000) def _wait_next_millis(self) - int: 等待下一毫秒 timestamp self._current_time() while timestamp self.last_timestamp: time.sleep(0.001) timestamp self._current_time() return timestamp # 使用示例 if __name__ __main__: # 初始化生成器machine_id 需全局唯一 generator SnowflakeGenerator(machine_id1) # 生成ID snowflake_id generator.generate() print(f生成的ID: {snowflake_id}) # 输出如 1318958301703372800 # 解析ID print(解析结果:, generator.parse(snowflake_id)) # 输出: {timestamp: 1715586304000, machine_id: 1, sequence: 0, datetime: 2024-05-12 14:25:04}
