首页 > 编程语言 >详细分析Python模块中的雪花算法(附模板)

详细分析Python模块中的雪花算法(附模板)

时间:2024-03-19 23:34:45浏览次数:32  
标签:datacenter Python timestamp self ID Snowflake 详细分析 id 模板

目录

前言

分布式ID的生成推荐阅读:分布式ID生成方法的超详细分析(全)

1. 基本知识

Snowflake 算法是一种用于生成全局唯一 ID 的分布式算法,最初由 Twitter 设计并开源

它被设计用于解决分布式系统中生成唯一 ID 的需求,特别是在微服务架构和大规模分布式系统中

Snowflake 算法的核心思想是利用时间戳、工作机器 ID 和序列号来生成全局唯一的 64 位长整型 ID

其核心的组成部分如下:

  • 时间戳(Timestamp):通常以毫秒为单位,用于标识生成 ID 的时间点。时间戳的精度对 ID 的唯一性至关重要

  • 工作机器 ID(Worker ID):用于标识不同的工作机器或节点。在分布式系统中,每个节点需要有唯一的标识

  • 数据中心 ID(Datacenter ID):用于标识不同的数据中心。在大规模分布式系统中,可能存在多个数据中心,每个数据中心需要有唯一的标识

  • 序列号(Sequence Number):用于解决同一时间戳下生成多个 ID 的冲突问题。序列号通常是一个自增的数字,通过与一定的位数掩码进行位运算来确保不会溢出

Snowflake 算法的作用:

  • 生成全局唯一 ID:Snowflake 算法可以在分布式系统中生成全局唯一的 ID,确保不同节点生成的 ID 不会冲突

  • 适用于分布式环境:由于Snowflake算法只依赖于机器的时钟和网络通信,因此非常适合在分布式环境中使用

  • 简单且高效:Snowflake 算法的实现相对简单,且性能高效,可以快速生成唯一 ID

2. 模板

以下模板带有注释

实现了一个 Snowflake 类,通过调用 generate 方法可以生成唯一的 Snowflake ID

Snowflake ID 是一个 64 位长整型,包含了时间戳、数据中心 ID、工作机器 ID 和序列号等信息

import time

class SnowFlake(object):
    def __init__(self, worker_id, datacenter_id, sequence=0):
        self.worker_id = worker_id  # 用于标识不同的工作机器
        self.datacenter_id = datacenter_id  # 用于标识不同的数据中心
        self.sequence = sequence  # 序列号,用于解决并发生成的 ID 冲突
        self.tw_epoch = 1288834974657  # Twitter Snowflake epoch (in milliseconds),Snowflake 算法的起始时间点
        
        # Bit lengths,用于计算位数
        self.worker_id_bits = 5  # 5位,最大值为31
        self.datacenter_id_bits = 5  # 5位,最大值为31
        self.max_worker_id = -1 ^ (-1 << self.worker_id_bits)  # 最大工作机器 ID
        self.max_datacenter_id = -1 ^ (-1 << self.datacenter_id_bits)  # 最大数据中心 ID
        self.sequence_bits = 12  # 12位,支持的最大序列号数
        self.sequence_mask = -1 ^ (-1 << self.sequence_bits)  # 序列号掩码,用于生成序列号
        
        # Create initial timestamp,初始化上一次生成 ID 的时间戳
        self.last_timestamp = self.current_timestamp()
        
        # Check worker_id and datacenter_id values,检查工作机器 ID 和数据中心 ID 的取值范围
        if self.worker_id > self.max_worker_id or self.worker_id < 0:
            raise ValueError(f"Worker ID must be between 0 and {self.max_worker_id}")
        if self.datacenter_id > self.max_datacenter_id or self.datacenter_id < 0:
            raise ValueError(f"Datacenter ID must be between 0 and {self.max_datacenter_id}")
    
    @staticmethod
    def current_timestamp():
        return int(time.time() * 1000)  # 获取当前时间戳,单位为毫秒
    
    def generate(self):
        timestamp = self.current_timestamp()  # 获取当前时间戳
        
        if timestamp < self.last_timestamp:  # 如果当前时间戳小于上一次生成 ID 的时间戳
            raise ValueError("Clock moved backwards. Refusing to generate ID for {} milliseconds".format(
                self.last_timestamp - timestamp))  # 抛出异常,时钟回拨
        
        if timestamp == self.last_timestamp:  # 如果当前时间戳等于上一次生成 ID 的时间戳
            self.sequence = (self.sequence + 1) & self.sequence_mask  # 增加序列号,并与序列号掩码进行与运算,防止溢出
            if self.sequence == 0:  # 如果序列号归零
                timestamp = self.wait_next_millis(self.last_timestamp)  # 等待下一毫秒
        else:
            self.sequence = 0  # 时间戳变化,序列号重置为零
        
        self.last_timestamp = timestamp  # 更新上一次生成 ID 的时间戳
        
        # Generate Snowflake ID,生成 Snowflake ID
        _id = ((timestamp - self.tw_epoch) << (self.worker_id_bits + self.datacenter_id_bits)) | (
                self.datacenter_id << self.worker_id_bits) | self.worker_id << self.sequence_bits | self.sequence  # 使用时间戳、数据中心 ID、工作机器 ID 和序列号生成 ID
        return f"{_id:016d}"  # 返回 64 位长整型 ID 的字符串表示,补齐到16位长度
    
    def wait_next_millis(self, last_timestamp):
        timestamp = self.current_timestamp()  # 获取当前时间戳
        while timestamp <= last_timestamp:  # 循环直到获取到下一毫秒的时间戳
            timestamp = self.current_timestamp()
        return timestamp  # 返回下一毫秒的时间戳

3. Demo

结合以上模板,放一个调用的过程:

# 示例用法
if __name__ == "__main__":
    # 假设有两个数据中心,每个数据中心有两个工作机器
    worker_id = 1
    datacenter_id = 1
    
    snowflake = SnowFlake(worker_id, datacenter_id)
    
    # 生成10个ID
    for i in range(10):
        print(snowflake.generate())

截图如下:

在这里插入图片描述

标签:datacenter,Python,timestamp,self,ID,Snowflake,详细分析,id,模板
From: https://blog.csdn.net/weixin_47872288/article/details/136858088

相关文章

  • 【Python】使用selenium对Poe批量模拟注册脚本
    配置好接码api即可实现自动化注册登录试用一体。运行后会注册账号并绑定邮箱与手机号进行登录试用。测试结果30秒一个号importreimporttimeimportrequestsfrombs4importBeautifulSoupfromseleniumimportwebdriverfromselenium.webdriver.chrome.option......
  • Python 机器学习 HMM模型三种经典问题
    ​ 隐马尔可夫模型(HiddenMarkovModel,HMM)是一个强大的工具,用于模拟具有隐藏状态的时间序列数据。HMM广泛应用于多个领域,如语音识别、自然语言处理和生物信息学等。在处理HMM时,主要集中于三个经典问题:评估问题、解码问题和学习问题。三个问题构成了使用隐马尔可夫模型时的基础......
  • 一文了解Python中的运算
    Python的运算符和其他语言类似数学运算>>>print 1+9        # 加法>>>print 1.3-4      # 减法>>>print 3*5        # 乘法>>>print 4.5/1.5    # 除法>>>print 3**2       # 乘方     >>>print 10%3      ......
  • Python小白的福利之基本数据类型
    简单的数据类型以及赋值变量不需要声明Python的变量不需要声明,你可以直接输入:>>>a = 10那么你的内存里就有了一个变量a,它的值是10,它的类型是integer(整数)。在此之前你不需要做什么特别的声明,而数据类型是Python自动决定的。>>>print a>>>print type(a)那......
  • Python 数据持久层ORM框架 TorToise模块(异步)
    文章目录TortoiseORM简介TortoiseORM特性TortoiseORM安装TortoiseORM数据库支持TortoiseORM创建模型aerich迁移工具简介aerich迁移工具安装aerich迁移工具使用TrotoiseORM查询数据TrotoiseORM修改数据TrotoiseORM删除数据TrotoiseORM新增数据......
  • 【蓝桥杯选拔赛真题70】python最短路径和 第十五届青少年组蓝桥杯python选拔赛真题 算
    目录python最短路径和一、题目要求1、编程实现2、输入输出二、算法分析三、程序编写四、程序说明五、运行结果六、考点分析七、 推荐资料1、蓝桥杯比赛2、考级资料3、其它资料python最短路径和第十五届蓝桥杯青少年组python比赛选拔赛真题一、题目要求(注:i......
  • Python的优点和缺点(详细解说版本)——《跟老吕学Python编程》附录资料
    Python的优点和缺点(详细解说版本)——《跟老吕学Python编程》附录资料Python的优点和缺点Python的优点Python语法简单Python是开源的程序员使用Python编写的代码是开源的。Python解释器和模块是开源的。Python是免费的Python是高级语言Python是解释型语言,能跨平......
  • Python面向对象编程之多态你学会了吗?
    在Python面向对象编程中,多态是一个非常重要的概念。多态意味着一个接口可以有多种实现方式,或者说一个接口可以被多种对象所实现。这在编程中非常重要,因为它可以帮助我们编写更加灵活和可扩展的代码。想象一下,如果你有一个函数,它需要处理不同的对象,但是这些对象都实现了同一......
  • Python面向对象编程之多继承,你真的懂了吗?
    hi,大家好!今天我们来聊聊Python面向对象编程中的一个重要概念——多继承!如果你还没搞清楚这个概念,那就赶紧跟着我一起学习吧!首先,我们来了解一下什么是继承。在面向对象编程中,继承是一个让子类可以继承父类的属性和方法的机制。这样,我们就可以避免重复编写相同的代码,并且让代......
  • python27安装pygame
    参考:https://cloud.tencent.com/developer/article/2089701我安装的是1.9.3版本https://pypi.org/project/pygame/1.9.3/#files按照自己本地的环境下载,比如我的是python27,windows64,我安装的就是 pygame-1.9.3-cp27-cp27m-win_amd64.whl安装命令:pipinstallxxxx.whl 试......