如何在Python中生成一个按时间排序的uid?

i1icjdpr  于 2022-11-19  发布在  Python
关注(0)|答案(4)|浏览(150)

这可能吗?我听说 cassandra 也有类似的东西:https://datastax.github.io/python-driver/api/cassandra/util.html
我一直在使用ISO timestampuuid4的连接,但结果太大(58个字符),可能是矫枉过正。

在我的上下文中保留序号不起作用(DynamoDB NoSQL)

值得注意的是,对于我的应用程序来说,只要uid不崩溃,在同一秒内批量创建的项是否以随机顺序排列并不重要。
我对最大长度没有具体的限制,理想情况下我希望看到不同长度的不同碰撞机会,但它需要小于58(我最初的尝试)
这将与DynamoDB(NoSQL数据库)一起用作排序键

fkaflof6

fkaflof61#

为什么uuid.uuid1不是连续的

uuid.uuid1(node=None, clock_seq=None)实际上是:

  • 60位时间戳(表示1582-10-15 00:00:00之后的100 ns间隔数)
  • 14位“时钟序列”
  • 48位“节点信息”(从网卡的mac地址或主机名或RNG生成)。

如果不提供任何参数,则调用System函数生成uuid。在这种情况下:

  • 目前还不清楚“时钟序列”是顺序的还是随机的。
  • 不清楚在多个进程中使用它是否安全(clock_seq是否可以在不同的进程中重复使用?)。在Python 3.7中,这个信息现在是可用的。

如果您提供clock_seqnode,则“使用纯Python实现”。在这种情况下,即使clock_seq为“固定值”:

  • 时间戳部分保证对当前进程中的所有调用都是连续的,即使在线程执行中也是如此。
  • clock_seq部分是随机生成的。但这并不重要,因为时间戳是连续且唯一的。
  • 对于多个进程来说,这是不安全的(如果在“相同的100-ns时间间隔”内调用具有相同clock_seq, nodeuuid1的进程,则可能返回冲突的值)

重用uuid.uuid1的解决方案

很容易看出,你可以通过提供clock_seqnode参数(使用python实现)来使uuid1序列化。

import time

from uuid import uuid1, getnode

_my_clock_seq = getrandbits(14)
_my_node = getnode()

def sequential_uuid(node=None):
    return uuid1(node=node, clock_seq=_my_clock_seq)
    # .hex attribute of this value is 32-characters long string

def alt_sequential_uuid(clock_seq=None):
    return uuid1(node=_my_node, clock_seq=clock_seq)

if __name__ == '__main__':
    from itertools import count
    old_n = uuid1()  # "Native"
    old_s = sequential_uuid()  # Sequential

    native_conflict_index = None

    t_0 = time.time()

    for x in count():
        new_n = uuid1()
        new_s = sequential_uuid()

        if old_n > new_n and not native_conflict_index:
            native_conflict_index = x

        if old_s >= new_s:
            print("OOops: non-sequential results for `sequential_uuid()`")
            break

        if (x >= 10*0x3fff and time.time() - t_0 > 30) or (native_conflict_index and x > 2*native_conflict_index):
            print('No issues for `sequential_uuid()`')
            break

        old_n = new_n
        old_s = new_s

    print(f'Conflicts for `uuid.uuid1()`: {bool(native_conflict_index)}')

多个进程问题

但是如果您在同一台计算机上运行一些并行进程,则:

  • 默认为uuid.get_node()node对于所有进程都是相同的;
  • 对于某些进程,clock_seq有很小的机会相同(机会为1/16384)

这可能会导致冲突!这是在同一台机器上的并行进程中使用uuid.uuid1时的常见问题,除非您可以从Python3.7访问SafeUUID。
如果您还确保为运行此代码的每个并行进程将node设置为唯一值,则应该不会发生冲突。
即使您使用SafeUUID并设置了唯一的node,如果它们是在不同的进程中生成的,那么仍然有可能具有非顺序的(但唯一的)id。
如果一些与锁相关的开销是可以接受的,那么可以将clock_seq存储在一些外部原子存储中(例如在"locked"文件中),并随着每次调用而递增:这允许所有并行进程上的node具有相同的值,并且还将使id-s连续。对于所有并行进程都是使用multiprocessing创建的子进程的情况:可以使用multiprocessing.Value“共享”clock_seq
因此,您始终必须记住:

  • 如果在同一台计算机上运行多个进程,则必须:
  • 确保node的唯一性.此问题为:您不能确保在相同的100 ns间隔内生成来自不同进程的连续id。但这是在进程启动时执行一次的非常“轻”的操作,通过以下方式实现:向默认节点“添加”某些内容,例如int(time.time()*1e9) - 0x118494406d1cc000,或者通过添加来自机器级原子数据库的某个计数器。
  • 确保“机器级原子clock_seq“和同一台机器上所有进程的相同node。这样,您将有一些开销用于“锁定”clock_seq,但可以保证id是连续的,即使在相同的100 ns间隔内在不同的进程中生成(除非您从同一进程中的多个线程调用uuid)。
  • 对于不同计算机上的进程:
  • 要么你必须使用一些“全球柜台服务”;
  • 或者不可能在相同的100纳秒间隔期间在不同的机器上生成连续的ID。

减小id的大小

生成UUID的一般方法非常简单,因此很容易从头开始实现类似的操作,例如,node_info部分使用较少的位:

import time
from random import getrandbits

_my_clock_seq = getrandbits(14)
_last_timestamp_part = 0
_used_clock_seq = 0

timestamp_multiplier = 1e7  # I'd recommend to use this value

# Next values are enough up to year 2116:
if timestamp_multiplier == 1e9:
    time_bits = 62  # Up to year 2116, also reduces chances for non-sequential id-s generated in different processes
elif timestamp_multiplier == 1e8:
    time_bits = 60  # up to year 2335
elif timestamp_multiplier == 1e7:
    time_bits = 56  # Up to year 2198.
else:
    raise ValueError('Please calculate and set time_bits')

time_mask = 2**time_bits - 1

seq_bits = 16
seq_mask = 2**seq_bits - 1

node_bits = 12
node_mask = 2**node_bits - 1

max_hex_len = len(hex(2**(node_bits+seq_bits+time_bits) - 1)) - 2  # 21

_default_node_number = getrandbits(node_bits)  # or `uuid.getnode() & node_mask`

def sequential_uuid(node_number=None):
    """Return 21-characters long hex string that is sequential and unique for each call in current process.

    Results from different processes may "overlap" but are guaranteed to
    be unique if `node_number` is different in each process.

    """
    global _my_clock_seq
    global _last_timestamp_part
    global _used_clock_seq
    if node_number is None:
        node_number = _default_node_number
    if not 0 <= node_number <= node_mask:
        raise ValueError("Node number out of range")

    timestamp_part = int(time.time() * timestamp_multiplier) & time_mask
    _my_clock_seq = (_my_clock_seq + 1) & seq_mask

    if _last_timestamp_part >= timestamp_part:
        timestamp_part = _last_timestamp_part
        if _used_clock_seq == _my_clock_seq:
            timestamp_part = (timestamp_part + 1) & time_mask
    else:
        _used_clock_seq = _my_clock_seq

    _last_timestamp_part = timestamp_part

    return hex(
        (timestamp_part << (node_bits+seq_bits))
        |
        (_my_clock_seq << (node_bits))
        |
        node_number
    )[2:]

备注:

  • 也许在数据库中简单地存储整数值(而不是十六进制字符串)会更好
  • 如果你将它存储为text/char,那么最好将integer转换为base64字符串,而不是转换为hex-string,这样会更短(21个字符的hex-string → 16个字符的b64编码字符串):
from base64 import b64encode

total_bits = time_bits+seq_bits+node_bits
total_bytes = total_bits // 8 + 1 * bool(total_bits % 8)

def int_to_b64(int_value):
    return b64encode(int_value.to_bytes(total_bytes, 'big'))

碰撞几率

  • 单进程:不可能发生冲突
  • 多个进程,每个进程中手工设置唯一clock_seq唯一node:不可能发生冲突
  • 随机设置node的多个进程(48位,时间“固定”):
  • 在几个过程中发生node碰撞的机会:
  • 10000个过程中有2个:约0.000018%
  • 100000个过程中有2个:0.0018%的百分比
  • 在2个进程中每秒有一次id与“冲突”node冲突的机会:
  • 对于“时间戳”间隔为100 ns(默认值为uuid.uuid1,在我代码中为timestamp_multiplier == 1e7):与3.72e-19 * avg_call_frequency²成比例
  • 对于10纳秒(timestamp_multiplier == 1e8)“时间戳”间隔:与3.72e-21 * avg_call_frequency²成比例
cigdeys3

cigdeys32#

在您链接的文章中,cassandra.util.uuid_from_time(time_arg,node=None,clock_seq=None)[source]似乎正是您要查找的内容。

def uuid_from_time(time_arg, node=None, clock_seq=None):
    """
    Converts a datetime or timestamp to a type 1 :class:`uuid.UUID`.

    :param time_arg:
      The time to use for the timestamp portion of the UUID.
      This can either be a :class:`datetime` object or a timestamp
      in seconds (as returned from :meth:`time.time()`).
    :type datetime: :class:`datetime` or timestamp

    :param node:
      None integer for the UUID (up to 48 bits). If not specified, this
      field is randomized.
    :type node: long

    :param clock_seq:
      Clock sequence field for the UUID (up to 14 bits). If not specified,
      a random sequence is generated.
    :type clock_seq: int

    :rtype: :class:`uuid.UUID`

    """
    if hasattr(time_arg, 'utctimetuple'):
        seconds = int(calendar.timegm(time_arg.utctimetuple()))
        microseconds = (seconds * 1e6) + time_arg.time().microsecond
    else:
        microseconds = int(time_arg * 1e6)

    # 0x01b21dd213814000 is the number of 100-ns intervals between the
    # UUID epoch 1582-10-15 00:00:00 and the Unix epoch 1970-01-01 00:00:00.
    intervals = int(microseconds * 10) + 0x01b21dd213814000

    time_low = intervals & 0xffffffff
    time_mid = (intervals >> 32) & 0xffff
    time_hi_version = (intervals >> 48) & 0x0fff

    if clock_seq is None:
        clock_seq = random.getrandbits(14)
    else:
        if clock_seq > 0x3fff:
            raise ValueError('clock_seq is out of range (need a 14-bit value)')

    clock_seq_low = clock_seq & 0xff
    clock_seq_hi_variant = 0x80 | ((clock_seq >> 8) & 0x3f)

    if node is None:
        node = random.getrandbits(48)

    return uuid.UUID(fields=(time_low, time_mid, time_hi_version,
                             clock_seq_hi_variant, clock_seq_low, node), version=1)

cassandra 并没有特别针对第一类UUID...

cclgggtu

cclgggtu3#

您应该能够用32位编码精确到秒的时间戳,时间范围为135年。这将只需要8个字符来表示十六进制。添加到uuid的十六进制表示(32个十六进制字符),将总共只有40个十六进制字符。
以这种方式编码时间戳需要选择一个基准年(例如2000年),并计算到当前日期(时间戳)的天数。将此天数乘以86400,然后加上从午夜开始的秒数。这样给予的值将小于2^32,直到达到2135年。

  • 请注意,您必须在时间戳前缀的十六进制编码形式中保留前导零,以便字母数字排序保留时间顺序。*

如果时间戳中多了几位,就可以增加时间范围和/或精度。如果多了8位(两个十六进制字符),就可以达到270年,精度达到百分之一秒。
请注意,您不必以10为基数来模拟秒的小数部分。对于相同的字符数,您可以将其分解为128而不是100,从而获得最佳的位使用率。在年份范围加倍的情况下,这仍然适合8位(2个十六进制字符)
在时间精确度(例如每秒或每100或128秒)内的恩怨几率是由uuid的范围所决定,因此对于所选的精确度而言,恩怨几率为1/2^128。增加时间戳记的精确度对减少恩怨几率的影响最大。它也是对索引键总大小影响最小的因素。

更有效的字符编码:27至29个字符键

您可以通过以base 64而不是16进行编码来显著减小密钥的大小,这样会给予27到29个字符(取决于您选择的精度)
请注意,对于时间戳部分,您需要使用一个编码函数,该函数将整数作为输入,并保留数字字符的排序序列。
例如:

def encode64(number, size):
    chars  = "+-0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"
    result = list()
    for _ in range(size):
        result.append(chars[number%64])
        number //= 64
    return "".join(reversed(result))

a = encode64(1234567890,6) # '-7ZU9G'
b = encode64(9876543210,6) # '7Ag-Pe'
print(a < b)               # True

u = encode64(int(uuid.uuid4()),22) # '1QA2LtMg30ztnugxaokVMk'

key = a+u  # '-7ZU9G1QA2LtMg30ztnugxaokVMk'  (28 characters)

您可以在编码前将时间戳记和uuid结合成单一数字,而不是将两个编码值串连起来,以保存更多的字符。
encode 64()函数每6位需要一个字符。
所以,135年来精确到秒:(32+128)/6 = 26.7 --〉27个字符
而不是(32/6 = 5.3 --〉6)+(128/6 = 21.3 --〉22)==〉28个字符

uid       = uuid.uuid4()
timeStamp = daysSince2000 * 86400 + int(secondsSinceMidnight)
key       = encode64( timeStamp<<128 | int(uid) ,27)

以270年的时间跨度和128分之一秒的精度:(40+128)/6 =28个字符

uid       = uuid.uuid4()
timeStamp = daysSince2000 * 86400 + int(secondsSinceMidnight)
precision = 128
timeStamp = timeStamp * precision + int(factionOfSecond * precision)
key       = encode64( timeStamp<<128 | int(uid) ,28)

使用29个字符,可以将精度提高到1024秒,将年份范围提高到2160年

UUID屏蔽:17至19个字符的按键

为了提高效率,你可以去掉uuid的前64位(它已经是一个时间戳了),然后把它和你自己的时间戳组合在一起,这样你给予的密钥的长度就可以是17到19个字符,而实际上不会损失冲突避免(取决于你选择的精度)。

mask  = (1<<64)-1
key   = encode64( timeStamp<<64 | (int(uid) & mask) ,19)

整数/数字键?

最后要注意的是,如果数据库支持非常大的整数或数字字段(140位或更多)作为键,则不必将组合数字转换为字符串。只需将其直接用作键即可。timeStamp<<128 | int(uid)的数字序列将遵循时间顺序。

ehxuflar

ehxuflar4#

The uuid6模块(pip install uuid6)解决了这个问题,它的目的是实现一个新的uuid变体标准的相应草案,参见here
示例代码:

import uuid6

for i in range(0, 30):
    u = uuid6.uuid7()
    print(u)
    time.sleep(0.1)

软件包建议使用uuid6.uuid7()
如果可能的话,实现应该使用UUID版本7,而不是UUID版本1和6。
UUID版本7具有一个时间顺序值字段,该字段源自广泛实现且广为人知的Unix纪元时间戳源,自1970年1月1日午夜以来的毫秒秒数(不包括闰秒)。

相关问题