DynamoDB批量更新咨询：并发请求下操作顺序一致性解决方案

针对你遇到的DynamoDB中字段变更后同步其他条目，以及并发场景下顺序一致性的问题，以下是几个可行的解决方案：

方案1：乐观锁（版本号）+ 事务批量更新 #

为每个需要同步的记录添加一个version数值字段，每次更新时基于当前版本号做条件校验，确保只有预期的版本才能被修改，从而避免并发覆盖。

步骤： #

1. 查询所有需要同步更新的记录，获取它们的主键和当前version值。
2. 启动DynamoDB事务，对每条记录执行UpdateItem操作，带上条件表达式version = :curr_version，同时将version自增1，更新目标index字段。
3. 如果事务执行失败（比如版本不匹配），说明存在并发修改，触发重试逻辑（可限制重试次数）。

代码示例（Python/boto3）： #

import boto3

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('your-table-name')

def sync_index_updates(target_index_old, target_index_new):
    # 查询所有包含旧index的记录
    response = table.scan(
        FilterExpression='index = :old_idx',
        ExpressionAttributeValues={':old_idx': target_index_old}
    )
    items_to_update = response['Items']
    
    if not items_to_update:
        return
    
    # 构建事务更新请求
    transact_items = []
    for item in items_to_update:
        transact_items.append({
            'Update': {
                'TableName': 'your-table-name',
                'Key': {'id': item['id']},
                'UpdateExpression': 'SET index = :new_idx, version = version + :incr',
                'ConditionExpression': 'version = :curr_version',
                'ExpressionAttributeValues': {
                    ':new_idx': target_index_new,
                    ':incr': 1,
                    ':curr_version': item['version']
                }
            }
        })
    
    # 执行事务
    try:
        table.transact_write_items(TransactItems=transact_items)
        print("同步更新成功")
    except dynamodb.meta.client.exceptions.TransactionCanceledException as e:
        print(f"并发冲突，重试: {e}")
        # 这里可以添加重试逻辑，比如递归调用或用重试库

方案2：全局独占锁控制并发 #

创建一条专门的锁记录，利用DynamoDB的条件表达式实现全局独占锁，确保同一时间只有一个请求能执行同步更新操作，从根源上避免并发顺序问题。

步骤： #

1. 尝试获取锁：向锁表（或同一张表）写入一条主键为Lock#IndexSync的记录，条件是该记录不存在（或已释放），同时设置TTL字段防止死锁。
2. 获取锁成功后，执行所有同步更新操作。
3. 操作完成后，删除锁记录释放锁。

代码示例（Python/boto3）： #

import time
from datetime import datetime, timedelta

def acquire_lock():
    lock_table = dynamodb.Table('lock-table')
    lock_key = {'lock_id': 'IndexSync'}
    ttl = int((datetime.now() + timedelta(minutes=5)).timestamp())  # 5分钟超时
    
    try:
        lock_table.put_item(
            Item={**lock_key, 'locked': True, 'ttl': ttl},
            ConditionExpression='attribute_not_exists(locked) OR ttl < :now',
            ExpressionAttributeValues={':now': int(time.time())}
        )
        return True
    except dynamodb.meta.client.exceptions.ConditionalCheckFailedException:
        return False

def release_lock():
    lock_table = dynamodb.Table('lock-table')
    lock_table.delete_item(Key={'lock_id': 'IndexSync'})

def sync_index_with_lock(target_index_old, target_index_new):
    # 获取锁
    while not acquire_lock():
        time.sleep(0.5)  # 等待并重试
    
    try:
        # 执行同步更新逻辑（已加锁，无需事务）
        response = table.scan(
            FilterExpression='index = :old_idx',
            ExpressionAttributeValues={':old_idx': target_index_old}
        )
        items_to_update = response['Items']
        
        with table.batch_writer() as batch:
            for item in items_to_update:
                batch.update_item(
                    Key={'id': item['id']},
                    UpdateExpression='SET index = :new_idx',
                    ExpressionAttributeValues={':new_idx': target_index_new}
                )
        print("同步更新成功")
    finally:
        # 释放锁
        release_lock()

方案3：DynamoDB Streams + Lambda（最终一致性） #

如果业务允许最终一致性，可以用DynamoDB Streams捕获字段变更事件，触发Lambda异步处理同步更新。这种方式无需处理前端并发，由AWS服务保证同一分区内的事件按顺序处理。

关键点： #

1. 开启表的DynamoDB Streams，捕获MODIFY类型的事件。
2. Lambda函数监听Stream事件，当检测到index字段变更时，查询所有需要同步的记录并更新。
3. 为保证幂等，Lambda函数可以用事件ID作为去重依据，避免重复处理。

注意： #

这个方案是异步的，无法保证实时一致性，但能大幅降低并发冲突的概率，适合对一致性要求不是强实时的场景。

内容的提问来源于stack exchange，提问作者Serhii