高并发Lambda下如何判断AWS SQS最后一条消息已处理？

Lambda 02开10并发时，多个实例同时更新Parent的iteration字段会触发DynamoDB的竞态条件：多个实例读取到同一个iteration值，各自加1后写回，导致最终的iteration值被互相覆盖，实际递增次数少于消息处理次数，自然无法达到numberOfChild的数值。

1. 用DynamoDB原子递增确保计数准确 #

这是最简单高效的方案，直接用DynamoDB的ADD操作符做原子递增，完全避免竞态问题：

def update_parent_and_check_completion(parent_id):
    # 原子递增iteration字段
    update_response = dynamodb.update_item(
        TableName='ParentTable',
        Key={'id': parent_id},
        UpdateExpression='ADD iteration :inc',
        ExpressionAttributeValues={':inc': 1},
        ReturnValues='ALL_NEW'
    )
    current_iter = update_response['Attributes']['iteration']
    
    # 获取该Parent的总子消息数
    parent = dynamodb.get_item(
        TableName='ParentTable',
        Key={'id': parent_id}
    )
    total_children = parent['Item']['numberOfChild']
    
    # 判断是否处理完最后一条消息
    if current_iter == total_children:
        # 触发后续流程
        trigger_post_processing_workflow(parent_id)

ADD操作是原子性的，哪怕10个Lambda实例同时请求，DynamoDB也会保证iteration被正确累加，不会出现覆盖。

2. 乐观锁兜底（适合复杂条件场景） #

如果业务需要更严格的状态判断，比如必须基于特定的iteration值才能更新，可以用条件表达式实现乐观锁，冲突时重试：

def safe_update_parent_iteration(parent_id):
    while True:
        # 读取当前的iteration和总子数
        parent = dynamodb.get_item(
            TableName='ParentTable',
            Key={'id': parent_id}
        )
        current_iter = parent['Item']['iteration']
        total_children = parent['Item']['numberOfChild']
        
        try:
            # 仅当数据库中的iteration与读取值一致时，才执行递增
            update_response = dynamodb.update_item(
                TableName='ParentTable',
                Key={'id': parent_id},
                UpdateExpression='SET iteration = iteration + :inc',
                ConditionExpression='iteration = :current_val',
                ExpressionAttributeValues={
                    ':inc': 1,
                    ':current_val': current_iter
                },
                ReturnValues='ALL_NEW'
            )
            new_iter = update_response['Attributes']['iteration']
            break
        except dynamodb.exceptions.ConditionalCheckFailedException:
            # 条件不满足，说明有其他实例先更新，重试
            continue
    
    if new_iter == total_children:
        trigger_post_processing_workflow(parent_id)

这种方式能确保每次递增都基于最新状态，重试逻辑可处理并发冲突。

3. 基于Child状态统计完成时机（适合小批量场景） #

如果不想依赖Parent的iteration，可以直接统计已标记为treated的Child数量：

def process_child_and_check_completion(child_id, parent_id):
    # 幂等更新Child的treated属性，避免重试重复标记
    dynamodb.update_item(
        TableName='ChildTable',
        Key={'id': child_id},
        UpdateExpression='SET treated = :val',
        ConditionExpression='attribute_not_exists(treated) OR treated = :false_val',
        ExpressionAttributeValues={
            ':val': True,
            ':false_val': False
        }
    )
    
    # 统计已处理的Child数量
    count_result = dynamodb.query(
        TableName='ChildTable',
        KeyConditionExpression='parent_id = :pid',
        FilterExpression='treated = :true_val',
        ExpressionAttributeValues={
            ':pid': parent_id,
            ':true_val': True
        },
        Select='COUNT'
    )
    processed_count = count_result['Count']
    
    # 获取总子数
    parent = dynamodb.get_item(
        TableName='ParentTable',
        Key={'id': parent_id}
    )
    total_children = parent['Item']['numberOfChild']
    
    if processed_count == total_children:
        trigger_post_processing_workflow(parent_id)

注意：如果子消息数量很大，这种计数查询的成本较高，且DynamoDB查询结果可能有短暂的最终一致性延迟，仅适合小批量场景。

• 给Parent新增is_completed字段：当iteration达到numberOfChild时，先检查is_completed是否为false，仅在未完成时触发后续流程，并将is_completed设为true，避免多个并发实例重复触发后续流程。
• 处理SQS消息要保证幂等：比如Child的treated更新要加条件，仅在未处理状态时才更新，防止消息重试导致重复计数。

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