快捷導(dǎo)航

Django python雪花算法實現(xiàn)方式

更新時間：2024年09月21日 14:24:11 作者：貪玩的小金魚

在Django項目中添加自定義模塊或應(yīng)用來封裝雪花算法,步驟包括創(chuàng)建應(yīng)用,編寫算法實現(xiàn)代碼至utils.py文件,及配置settings.py,此方法可方便在項目中隨處調(diào)用雪花算法,適用于需要唯一ID生成的場景

Django python雪花算法實現(xiàn)

雪花算法封裝到 Django 項目中，創(chuàng)建一個自定義的模塊或應(yīng)用，并將算法實現(xiàn)放在其中。

在你的Django項目中創(chuàng)建一個新的應(yīng)用

（例如，命名為 snowflake）：

python manage.py startapp snowflake

在新應(yīng)用的目錄中創(chuàng)建一個utils.py文件

并在其中放置雪花算法的實現(xiàn)代碼：

# 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

將新創(chuàng)建的應(yīng)用添加到settings.py 的 INSTALLED_APPS 屬性中：

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

你可以在你的 Django 項目的任何地方使用雪花算法了。

# 在你的視圖或模型中使用雪花算法
#根據(jù)分布式修改 worker_id=1  防止高并發(fā)出現(xiàn)ID重復(fù)。
from snowflake.utils import Snowflake
def my_view(request):
    snowflake = Snowflake(worker_id=1)
    unique_id = snowflake.generate_id()
    print(unique_id)
    # 使用唯一 ID 執(zhí)行其他操作