如何为AWS Lambda写入DynamoDB的MQTT数据添加递增序列号？

问题背景 #

使用AWS Lambda编写Python程序，将ESP8266通过MQTT发送的timestamp和accurrent数据存入DynamoDB表esp8266_v2，希望为每条数据添加1、2、3...的递增序列号。尝试通过describe_table获取ItemCount计算序列号，但所有条目的序列号均为1，无法实现递增效果。

原方法失效原因 #

1. ItemCount是近似值：DynamoDB的ItemCount是后台异步统计的近似值，并非实时更新（通常每6小时刷新一次），新插入的数据不会立刻反映到这个数值中，导致刚创建的表或新插入数据后，ItemCount始终为0，计算出的序列号都是0+1=1。
2. 并发冲突问题：即使ItemCount是实时值，高并发场景下多个Lambda请求会同时读取到相同的ItemCount，最终生成重复的序列号，无法保证唯一性。

正确解决方案：使用DynamoDB原子计数器 #

DynamoDB的update_item操作支持原子性的数值递增，可以通过维护一个专门的计数器来生成唯一递增的序列号，完全避免上述问题。

步骤1：创建计数器表 #

在DynamoDB中创建一个名为counters的表，设置：

• 分区键：counter_name（字符串类型）
• 手动插入初始计数器项：
  • counter_name：esp8266_serial
  • current_value：0（数字类型）

步骤2：修改Lambda代码 #

使用原子递增操作获取序列号，再存入业务数据：

import boto3

def lambda_handler(event, context):
    # 初始化DynamoDB资源客户端
    dynamodb = boto3.resource('dynamodb')
    
    # 引用计数器表和业务数据表
    counter_table = dynamodb.Table('counters')
    data_table = dynamodb.Table('esp8266_v2')
    
    # 原子递增计数器，获取更新后的序列号
    counter_update = counter_table.update_item(
        Key={'counter_name': 'esp8266_serial'},
        UpdateExpression='SET current_value = current_value + :increment',
        ExpressionAttributeValues={':increment': 1},
        ReturnValues='UPDATED_NEW'  # 返回更新后的最新值
    )
    
    # 提取生成的序列号
    serial_number = counter_update['Attributes']['current_value']
    
    # 将数据存入业务表
    data_table.put_item(
        Item={
            'serial_number': serial_number,
            'timestamp': event['timestamp'],
            'accurrent': event['accurrent']
        }
    )
    
    return {
        'statusCode': 200,
        'body': f'序列号 {serial_number} 已成功添加。'
    }

方案优势 #

• 原子性：update_item的递增操作是原子执行的，即使多个请求同时触发，也不会生成重复的序列号。
• 实时性：每次请求都会立刻获取到最新的递增序列号，不存在延迟问题。
• 可扩展性：可以为不同的业务场景创建不同的计数器项，灵活复用。

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