Python代码示例：Twitter的Snowflake雪花算法代码示例-蒲公英云

Python代码示例：Twitter的Snowflake雪花算法代码示例

你的名字 2024-02-17 10:08 89阅读 0赞

以下是一个简化的示例代码，演示了如何实现Twitter的Snowflake算法：

import time
class Snowflake:
    def __init__(self, worker_id, datacenter_id, sequence=0):
        self.worker_id = worker_id
        self.datacenter_id = datacenter_id
        self.sequence = sequence
        self.last_timestamp = -1
    def generate_id(self):
        timestamp = self._get_timestamp()
        if timestamp < self.last_timestamp:
            raise Exception("Invalid system clock!")
        if timestamp == self.last_timestamp:
            self.sequence = (self.sequence + 1) & 4095
            if self.sequence == 0:
                timestamp = self._wait_next_millis(self.last_timestamp)
        else:
            self.sequence = 0
        self.last_timestamp = timestamp
        return (
            (timestamp << 22)
            | (self.datacenter_id << 17)
            | (self.worker_id << 12)
            | self.sequence
        )
    def _get_timestamp(self):
        return int(time.time() * 1000)
    def _wait_next_millis(self, last_timestamp):
        timestamp = self._get_timestamp()
        while timestamp <= last_timestamp:
            timestamp = self._get_timestamp()
        return timestamp

使用示例：

# 创建Snowflake实例，传入worker_id和datacenter_id
snowflake = Snowflake(worker_id=1, datacenter_id=1)
# 生成唯一ID
unique_id = snowflake.generate_id()
print(unique_id)