雪花算法生成 ID 詳解

作者：子午 2025-09-16 07:00:00

雪花算法（Snowflake）是一種由 Twitter 開源的分布式唯一ID生成算法，用于在高并發分布式系統下生成全局唯一、遞增的長整型ID。

一、介紹

雪花算法（Snowflake）是一種由 Twitter 開源的分布式唯一ID生成算法，用于在高并發分布式系統下生成全局唯一、遞增的長整型ID。

時間戳	數據中心ID	機器ID	序列號
41位	5位	5位	12位

具體含義：

第1位占用1bit，其值始終是0，可看做是符號位不使用。
41位時間戳：毫秒級，能用69年（2^41毫秒）
5位數據中心ID：支持32個數據中心
5位機器ID：支持每個數據中心32臺機器
12位序列號：每毫秒同一機器可生成4096個不同ID

最終組成64位整數，既保證唯一性，也能做到大致有序（遞增）。

二、與其他ID方案對比

方案	格式	唯一性	有序性	性能	適用場景
雪花算法	64位整型	很高（全局）	是（時間排序）	非常高（本地生成）	分布式系統
UUID	128位字符串	極高（全局）	否	較高（本地生成）	分布式、無序唯一
自增ID	整型	僅表內唯一	是	依賴DB/緩存	單機、簡單場景

三、雪花算法優勢

高性能、本地生成：不依賴數據庫或第三方服務，單臺可毫秒生成數千個ID
分布式唯一性強：不重復、可控
有序遞增：帶有時間戳，ID基本有序，有利于數據庫索引
存儲高效：64位long型，存儲和索引效率高（遠小于UUID）
可擴展性強：支持多機房、多節點并發

UUID的特點和劣勢：

高唯一性：全球唯一，沖突概率極低
無序、字符串型：UUID 不帶時間，不遞增，作為主鍵時插入、索引效率低
存儲空間大：128位，通常以36字符字符串存儲（如 550e8400-e29b-41d4-a716-446655440000）

自增ID劣勢：

分布式難協調，跨庫唯一性難保證
安全性差，容易被猜測（業務有安全要求時不推薦）
分布式難協調，跨庫唯一性難保證

四、場景案例

業務場景：

服務目前QPS10萬，預計幾年之內會發展到百萬。
當前機器三地部署，上海，北京，深圳都有。
當前機器10臺左右，預計未來會增加至百臺。

這個時候我們根據上面的場景可以再次合理的劃分64bit，QPS幾年之內會發展到百萬，那么每毫秒就是千級的請求，目前10臺機器那么每臺機器承擔百級的請求，為了保證擴展，后面的循環位可以限制到1024，也就是2^10，那么循環位10位就足夠了。機器三地部署我們可以用3bit總共8來表示機房位置，當前的機器10臺，為了保證擴展到百臺那么可以用7bit 128來表示，時間位依然是41bit,那么還剩下64-10-3-7-41-1 = 2bit，還剩下2bit可以用來進行擴展。

五、時鐘回撥

因為機器的原因會發生時間回撥，我們的雪花算法是強依賴我們的時間的，如果時間發生回撥，有可能會生成重復的ID。我們可以用當前時間和上一次的時間進行判斷，如果當前時間小于上一次的時間那么肯定是發生了回撥，算法會直接拋出異常。

六、代碼實現

示例代碼：

# snowflake.py
# Twitter's Snowflake algorithm implementation which is used to generate distributed IDs.
# https://github.com/twitter-archive/snowflake/blob/snowflake-2010/src/main/scala/com/twitter/service/snowflake/IdWorker.scala

import time

# 64位ID的劃分
WORKER_ID_BITS = 5
DATACENTER_ID_BITS = 5
SEQUENCE_BITS = 12

# 最大取值計算
MAX_WORKER_ID = -1 ^ (-1 << WORKER_ID_BITS)  # 2**5-1 0b11111
MAX_DATACENTER_ID = -1 ^ (-1 << DATACENTER_ID_BITS)

# 移位偏移計算
WOKER_ID_SHIFT = SEQUENCE_BITS
DATACENTER_ID_SHIFT = SEQUENCE_BITS + WORKER_ID_BITS
TIMESTAMP_LEFT_SHIFT = SEQUENCE_BITS + WORKER_ID_BITS + DATACENTER_ID_BITS

# 序號循環掩碼
SEQUENCE_MASK = -1 ^ (-1 << SEQUENCE_BITS)

# Twitter元年時間戳
TWEPOCH = 1288834974657


class ClockBackwardsException(Exception):
    """時鐘回撥異常"""

    def __init__(self, last_timestamp, current_timestamp):
        super().__init__(
            f"時鐘回撥異常：當前時間戳 {current_timestamp} 小于上次生成ID的時間戳 {last_timestamp}，系統拒絕生成ID！")
        self.last_timestamp = last_timestamp
        self.current_timestamp = current_timestamp


class Snowflake(object):
    """
    用于生成IDs
    """

    def __init__(self, datacenter_id, worker_id, sequence=0):
        """
        初始化
        :param datacenter_id: 數據中心（機器區域）ID
        :param worker_id: 機器ID
        :param sequence: 序號
        """
        # sanity check
        if worker_id > MAX_WORKER_ID or worker_id < 0:
            raise ValueError('worker_id值越界')

        if datacenter_id > MAX_DATACENTER_ID or datacenter_id < 0:
            raise ValueError('datacenter_id值越界')

        self.worker_id = worker_id
        self.datacenter_id = datacenter_id
        self.sequence = sequence

        self.last_timestamp = -1# 上次計算的時間戳

    def _gen_timestamp(self):
        """
        生成整數時間戳
        :return:int timestamp
        """
        return int(time.time() * 1000)

    def get_id(self):
        """
        獲取新ID
        :return:
        """
        timestamp = self._gen_timestamp()

        # 時鐘回撥
        if timestamp < self.last_timestamp:
            raise ClockBackwardsException(self.last_timestamp, timestamp)

        if timestamp == self.last_timestamp:
            self.sequence = (self.sequence + 1) & SEQUENCE_MASK
            if self.sequence == 0:
                timestamp = self._til_next_millis(self.last_timestamp)
        else:
            self.sequence = 0

        self.last_timestamp = timestamp

        new_id = ((timestamp - TWEPOCH) << TIMESTAMP_LEFT_SHIFT) | (self.datacenter_id << DATACENTER_ID_SHIFT) | \
                 (self.worker_id << WOKER_ID_SHIFT) | self.sequence
        return new_id

    def _til_next_millis(self, last_timestamp):
        """
        等到下一毫秒
        """
        timestamp = self._gen_timestamp()
        while timestamp <= last_timestamp:
            timestamp = self._gen_timestamp()
        return timestamp

# settings中配置數據中心 和 機器號
# 這個地方是寫死的，后續如果部署到服務器上，可以使用讀取環境變量的形式
# 數據中心標識、機器號標識
DATACENTER_ID = 0
WORKER_ID = 0

user.py模型中進行修改：

from . import Base
from sqlalchemy import Column, BigInteger
from utils.snowflake import Snowflake
from settings import DATACENTER_ID, WORKER_ID

snowflake = Snowflake(DATACENTER_ID, WORKER_ID)


def generate_snowflake_id():
    new_id = snowflake.get_id()
    return new_id


class User(Base):
    __tablename__ = 'user'
    id = Column(BigInteger, primary_key=True, default=generate_snowflake_id)

責任編輯：趙寧寧來源：子午Python

雪花算法 ID Python