Django python雪花算法实现方式

python manage.py startapp snowflake

# snowflake/utils.py
import time

class Snowflake:
    def __init__(self, worker_id, epoch=0):
        self.worker_id = worker_id
        self.epoch = epoch

        self.sequence = 0
        self.last_timestamp = -1

    def generate_id(self):
        timestamp = int(time.time() * 1000)

        if timestamp == self.last_timestamp:
            self.sequence = (self.sequence + 1) & 4095
            if self.sequence == 0:
                timestamp = self.wait_next_millis(timestamp)
        else:
            self.sequence = 0

        self.last_timestamp = timestamp

        return ((timestamp - self.epoch) << 22) | (self.worker_id << 12) | self.sequence

    def wait_next_millis(self, last_timestamp):
        timestamp = int(time.time() * 1000)
        while timestamp <= last_timestamp:
            timestamp = int(time.time() * 1000)
        return timestamp

# settings.py
INSTALLED_APPS = [
    ...
    'snowflake',
    ...
]

# 在你的视图或模型中使用雪花算法
#根据分布式修改 worker_id=1  防止高并发出现ID重复。
from snowflake.utils import Snowflake
def my_view(request):
    snowflake = Snowflake(worker_id=1)
    unique_id = snowflake.generate_id()
    print(unique_id)
    # 使用唯一 ID 执行其他操作