如何在Python中生成时序uid？

Question

这可能吗？我听说卡桑德拉也有类似的事情：https://datastax.github.io/python-driver/api/cassandra/util.html

我一直在使用一个ISO timestamp和一个uuid4相连，但是结果太大了（58个字符），而且可能会过度使用。在

在我的上下文中，保持一个序列号不起作用（DynamoDB NoSQL）

值得注意的是，对于我的应用程序来说，批处理/同一秒创建的项是否是随机顺序并不重要，只要uid不会崩溃。在

我对最大长度没有具体限制，理想情况下，我希望看到不同长度的不同碰撞概率，但它必须小于58（我最初的尝试）

这将与DynamoDB（NoSQL数据库）一起用作排序键

Answer 1

你应该能够在135年的时间范围内用32位编码精确到秒的时间戳。这将只需要8个字符来表示十六进制。添加到uuid的十六进制表示中（32个十六进制字符），总共只有40个十六进制字符。在

以这种方式编码时间戳需要选择一个基准年（例如2000年）并计算到当前日期为止的天数（时间戳）。用这个天数乘以86400，然后加上从午夜开始的秒数。这将为您提供小于2^32的值，直到您达到2135年。在

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

在时间戳中多加几位，就可以增加时间范围和/或精度。再加上8个比特（两个十六进制字符），你可以用最长270年的时间精确到百分之一秒。
请注意，您不必在以10为基数的范围内对秒数进行建模。对于相同数量的字符，将其分解为128个而不是100个，从而获得最佳的位使用率。随着年份范围的翻倍，它仍然适合8位（2个十六进制字符）

在时间精度（即每秒或每100或128秒）内的碰撞概率由uuid的范围决定，因此对于选定的精度，碰撞概率将为1/2^128。提高时间戳的精度对减少碰撞概率的影响最大。它也是对密钥总大小影响最小的因素。在

更高效的字符编码：27到29个字符键

您可以通过将其编码为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合并为一个数字，而不是将两个编码的值串联起来，从而节省更多的字符。在

encode64（）函数每6位需要一个字符。在

因此，135年来精确到第二位：（32+128）/6=26.7>；27个字符

而不是（32/6=5.3>；6）+（128/6=21.3>；22）==>；28个字符

^{pr2}$

有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)的数字序列将遵循时间顺序。在

Answer 2

在您链接的文章中，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)

没有什么卡桑德拉是针对1型UUID的。。。在

Answer 3

为什么uuid.uuid1不是连续的

uuid.uuid1(node=None, clock_seq=None)有效：

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

若不提供任何参数，则调用系统函数生成uuid。在这种情况下：

• 目前还不清楚“时钟序列”是顺序的还是随机的。在
• 不清楚在多个进程中使用它是否安全（是否可以在不同的进程中重复clock_seq）。在python3.7中this info is now available。在

如果您提供clock_seq或node，那么“使用纯python实现”。在这种情况下，即使是clock_seq的“固定值”：

• 时间戳部分是当前进程中所有调用的guaranteed to be sequential，即使在线程执行中也是如此。在
• clock_seq部分是随机生成的。但这并不是关键的annymore，因为时间戳是连续的和唯一的。在
• 它对于多个进程是不安全的（使用相同的clock_seq, node调用uuid1的进程，如果在“相同的100ns时间间隔”内调用，则可能返回冲突的值）

重用uuid.uuid1的解决方案

很容易看出，您可以通过提供clock_seq或{}参数（使用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)}')

多流程问题

但是如果在同一台机器上运行一些并行进程，则：

• node默认为uuid.get_node()对所有进程都是相同的
• clock_seq对于某些进程来说，几乎没有相同的机会（几率为1/16384）

那可能会导致冲突！这是使用 uuid.uuid1在同一台机器上的并行进程中，除非您可以从Python3.7访问{a3}。在

如果您确保还为运行此代码的每个并行进程将node设置为唯一值，则不应发生冲突。在

即使您使用的是SafeUUID，并设置了unique node，如果它们是在不同的进程中生成的，那么仍然可以有非连续的（但唯一的）id。在

如果一些与锁相关的开销是可接受的，那么您可以将clock_seq存储在某些外部原子存储器中（例如在"locked"文件中），并在每次调用时递增它：这允许在所有并行进程上对node具有相同的值，并且还会使id-s连续。对于所有并行进程都是使用multiprocessing创建的子进程的情况：clock_seq可以使用multiprocessing.Value来“共享”

因此，您必须始终记住：

• 如果在同一台计算机上运行多个进程，则必须：

  • 确保node的唯一性。这个解决方案的问题是：您不能确定在相同的100ns间隔内从不同的进程中生成顺序id。但这是一个非常“轻”的操作，在进程启动时执行一次，并通过“添加”一些东西到默认节点，例如int(time.time()*1e9) - 0x118494406d1cc000，或者通过从计算机级原子数据库中添加一些计数器来实现。

  • 确保“机器级原子clock_seq”和同一台机器上所有进程的node。这样，您将有一些“锁定”clock_seq的开销，但是即使在相同的100ns间隔内在不同的进程中生成id-s，id-s也保证是连续的（除非您是从同一进程中的多个线程调用uuid）。

• 对于不同机器上的进程：

  • 要么你必须使用一些“全球柜台服务”；

  • 在同一时间间隔内，在相同的机器上可能生成不同的ID。

缩小id的大小

{30的方法很容易实现{30位，因此从scratch生成id很容易}部件：

^{pr2}$

注意事项：

• 也许最好在数据库中简单地存储整数值（而不是十六进制字符串）
• 如果将其存储为text/char，则最好将整数转换为base64字符串，而不是将其转换为十六进制字符串。这样它就会更短（21个字符的十六进制字符串→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%
    • 十万分之二的过程中：0.0018%

  • 每秒在2个进程中与“碰撞”node发生单次碰撞的几率：

    • 对于“时间戳”间隔为100ns（默认值为uuid.uuid1，在我的代码中，timestamp_multiplier == 1e7）：与3.72e-19 * avg_call_frequency²成比例

    • 对于10 ns（timestamp_multiplier == 1e8）的“时间戳”间隔：与3.72e-21 * avg_call_frequency²