DynamoDB 实现自增 ID 的两种方法_石家庄创泽智能科技有限公司

新闻中心 NEWS CENTER

您当前位置：首页 > 新闻中心 > 网络学院

DynamoDB 实现自增 ID 的两种方法

2025-11-22

浏览次数：次

返回列表

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 都需要一次额外的写入操作来更新计数器。

适用场景：

韩国商城购物MORNING MALL

主要增加论坛整合，在后台内置网银，快钱支付宝等实时在线支付平台支付宝支付方式改成在收银台统一支付并且修改了收到已付款定单后台显示定单确认功能[这功能非常强大，自动确认] 并且增加了商城内短信功能，商城店主可以自由与会员之间实时交谈。改正给ID添加积分后，登陆到前台，在 MEMBER LOGIN 下面的积分仍然显示为0的问题修改订单确认中投递&包装方法没有根据前面的选择而改

0 查看详情韩国商城购物MORNING MALL

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