Python利用雪花算法实现生成唯一ID

Python利用雪花算法实现生成唯一ID

雪花算法简介

雪花算法也叫雪花ID，是以Twitter的Snowflake算法为基础而开发出来的。雪花算法可以生成唯一ID，且有一定的顺序性，适用于分布式系统中的ID生成。

实现原理

雪花ID是64位的，其中第 1 个bit是符号位，始终为0；后41位为时间戳，单位是毫秒级，可以用约69年；接着的10位是机器 ID，可以部署在多台机器上，每台机器分配不同的ID；最后的12位是序列号，同毫秒内产生的ID序号，能够生成预计4个序列，最多4096个ID。

Python实现

下面将通过Python实现雪花算法生成唯一ID。

import time
import threading

# 定义全局变量为0
seq = 0
last_timestamp = -1

def gen_id():
    global seq
    global last_timestamp

    # 获取时间戳并转换为毫秒
    timestamp = int(time.time() * 1000)

    if last_timestamp > timestamp:
        raise ValueError("Clock moved backwards. Refusing to generate id")

    if last_timestamp == timestamp:
        seq = (seq + 1) & 0xFFF   # 4095

        if seq == 0:
            timestamp = til_next_millis(last_timestamp)
    else:
        seq = 0

    last_timestamp = timestamp

    # 偏移时间戳
    timestamp -= 1546300800000  # 2019-01-01

    return (timestamp << 22) | (0 << 12) | seq

def til_next_millis(last_timestamp):
    timestamp = int(time.time() * 1000)

    while timestamp <= last_timestamp:
        timestamp = int(time.time() * 1000)

    return timestamp

if __name__ == '__main__':
    num_threads=10
    def id_worker():
        thread_name = threading.currentThread().getName()
        for j in range(5):
            print("thread-{} id={}".format(thread_name, gen_id()))

    threads = []
    for i in range(num_threads):
        t = threading.Thread(target=id_worker)
        threads.append(t)

    for t in threads:
        t.start()

    for t in threads:
        t.join()

在上述代码中，代码中的def gen_id()函数实现了ID的生成。全局变量last_timestamp用来记录上次生成ID时的时间戳，全局变量seq用来记录序列号，两个变量用来实现顺序性要求。if判断语句用来保证ID在本机时间内的唯一性。最后，代码通过位运算，将时间戳向左移动22位，机器ID向左移动12位，序列号直接累加，将位运算结果合并返回生成的唯一ID。

示例说明

下面给出两个示例说明。

示例1

假设在分布式系统中，我们要求生成ID的总长度为20位，其中10位表示当前机器的编号，10位表示时间戳加序列号。

• 假设当前机器的编号为100，则机器ID为0000000110；
• 假设当前时间为2022年8月8日10点20分30秒，毫秒为500，则对应的时间戳为9447280220500；
• 假设当前时间戳下的序列号为50，则对应的序列号为000110010；

则该ID生成的结果为：11000000001110010050。

示例2

假设在分布式系统中，我们要求生成ID的总长度为10位。

• 假设当前机器的编号为50，则机器ID为000001100；
• 假设当前时间戳下的序列号为250，则对应的序列号为0111111010；

则该ID生成的结果为：00001100000111111010。

总结

通过Python实现雪花算法生成唯一ID，能实现自己ID的生成，以便分布式系统的使用。