DynamoDB 实现自增 ID 的两种方法-Python教程-PHP中文网

DynamoDB 实现自增 ID 的两种方法

碧海醫心

发布： 2025-11-22 09:44:02

原创

215人浏览过

dynamodb 实现自增 id 的两种方法

本文介绍了在 DynamoDB 中实现自增 ID 的两种常用方法，由于 DynamoDB 本身不支持像 MySQL 那样的原生自增功能，因此需要通过原子计数器或利用排序键的特性来实现。文章详细讲解了每种方法的实现原理、代码示例、优缺点以及适用场景，帮助开发者选择最适合自身业务需求方案。

在关系型数据库中，自增 ID 是一种常见的需求，可以方便地生成唯一且有序的标识符。然而，DynamoDB 作为 NoSQL 数据库，并不直接支持自增 ID 的功能。本文将介绍两种在 DynamoDB 中实现类似功能的常用方法。

方法一：使用原子计数器

原子计数器是一种利用 DynamoDB 的原子性操作来实现自增 ID 的方法。其核心思想是维护一个专门的计数器项，每次需要生成新的 ID 时，通过 UpdateItem 操作原子性地递增该计数器的值，并将递增后的值作为新的 ID。

实现步骤：

创建计数器项： 在 DynamoDB 表中创建一个特殊的项，用于存储计数器的值。例如，可以设置分区键为 'orderCounter'，属性名为 'count'。
原子递增计数器： 使用 UpdateItem 操作递增计数器的值，并要求 DynamoDB 返回更新后的值。

import boto3

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('orders')

# 原子递增计数器
response = table.update_item(
    Key={'pk': 'orderCounter'},
    UpdateExpression="ADD #cnt :val",
    ExpressionAttributeNames={'#cnt': 'count'},
    ExpressionAttributeValues={':val': 1},
    ReturnValues="UPDATED_NEW"
)

# 获取新的 ID 值
next_order_id = response['Attributes']['count']

# 使用新的 ID
table.put_item(
    Item={'pk': str(next_order_id), 'deliveryMethod': 'expedited'}
)

登录后复制

优点：

简单易懂： 实现逻辑清晰，易于理解和维护。
保证唯一性： DynamoDB 的原子性操作保证了生成的 ID 的唯一性。
无竞争条件： 所有对计数器项的写入都是串行执行的，避免了竞争条件的发生。

缺点：

吞吐量限制： 由于所有 ID 的生成都依赖于对同一个计数器项的更新，因此该方法的吞吐量受限于单个 DynamoDB 分区的最大吞吐量。
额外的写入成本： 每次生成新的 ID 都需要一次额外的写入操作来更新计数器。

适用场景：

微信二级防封域名

防封域名方法千千种，我们只做最简单且有用的这一种。微信域名防封是指通过技术手段来实现预付措施，一切说自己完全可以防封的那都是不可能的，一切说什么免死域名不会死的那也是吹牛逼的。我们正在做的是让我们的推广域名寿命更长一点，成本更低一点，效果更好一点。本源码采用 ASP+ACCESS 搭建，由于要用到二级域名，所以需要使用独享云虚机或者云服务器，不支持虚拟主机使用，不支持本地测试。目前这是免费测试版，

查看详情

对 ID 生成的吞吐量要求不高，例如，订单量较小的系统。
对 ID 的唯一性要求非常严格。

方法二：使用排序键

该方法利用 DynamoDB 表的主键中的排序键来实现自增 ID 的功能。其核心思想是利用排序键的排序特性，通过查询当前最大的排序键值，然后递增该值作为新的 ID。

实现步骤：

查询最大排序键值： 使用 Query 操作查询具有相同分区键的项集合中最大的排序键值。
递增排序键值： 将查询到的最大排序键值递增 1，作为新的 ID。
写入新项： 使用 PutItem 操作将包含新 ID 的项写入 DynamoDB 表。为了避免并发写入导致 ID 冲突，可以使用条件表达式 attribute_not_exists(pk) 来确保只有在具有相同分区键和排序键的项不存在时才写入。

import boto3
from boto3.dynamodb.conditions import Key

PROJECT_ID = 'projectA'

dynamodb = boto3.resource('dynamodb')
client = dynamodb.Table('projects')
highest_issue_id = 0
saved = False

# 查询最大排序键值
response = client.query(
    KeyConditionExpression=Key('pk').eq(PROJECT_ID),
    ScanIndexForward=False,
    Limit=1
)

# 获取最大排序键值
if response['Count'] > 0:
    highest_issue_id = int(response['Items'][0]['sk'])

while not saved:
    try:
        # 写入新项，并使用条件表达式避免并发写入
        response = client.put_item(
            Item={
                'pk': PROJECT_ID,
                'sk': highest_issue_id + 1,
                'priority': 'low'
            },
            ConditionExpression='attribute_not_exists(pk)'
        )
        saved = True
    # 如果条件表达式失败，则说明发生了并发写入，需要重新查询最大排序键值并重试
    except dynamodb.meta.client.exceptions.ConditionalCheckFailedException as e:
        highest_issue_id = highest_issue_id + 1

登录后复制

优点：