如何确认Lambda已完成指定客户的全部SQS消息处理并触发通知

针对你的场景——10个客户各1000条SQS消息由Lambda消费，需确认客户A的1000条处理完成后发送通知，以下是几种实用的落地方案：

方案一：DynamoDB计数器+原子更新（最可靠） #

这是最精准且能处理幂等性的方案，核心思路是用DynamoDB存储客户的消息处理状态，Lambda处理每条消息后更新计数，达到总数时触发通知。

具体步骤 #

1. 初始化状态记录
   在向SQS推送客户A的1000条消息前，先在DynamoDB中创建一条状态记录：

   {
     "customer_id": "A",
     "total_messages": 1000,
     "processed_count": 0,
     "status": "processing",
     "notified": false
   }
   

2. Lambda处理消息并更新计数
   Lambda每次处理完一条客户A的消息后，调用DynamoDB的UpdateItem做原子递增，同时检查是否完成：

   import boto3
   
   dynamodb = boto3.resource('dynamodb')
   table = dynamodb.Table('customer_message_status')
   
   def lambda_handler(event, context):
       for record in event['Records']:
           # 解析消息中的客户ID
           customer_id = record['body'].get('customer_id')
           if customer_id != 'A':
               continue
   
           # 原子更新处理计数，并获取更新后的值
           response = table.update_item(
               Key={'customer_id': customer_id},
               UpdateExpression='ADD processed_count :inc',
               ExpressionAttributeValues={':inc': 1},
               ReturnValues='ALL_NEW'
           )
           new_item = response['Attributes']
   
           # 检查是否完成且未发送过通知
           if new_item['processed_count'] == new_item['total_messages'] and not new_item['notified']:
               # 更新通知状态，避免重复发送
               table.update_item(
                   Key={'customer_id': customer_id},
                   UpdateExpression='SET notified = :val',
                   ExpressionAttributeValues={':val': True}
               )
               # 触发通知逻辑：邮件或WebSocket
               send_notification(customer_id)
   
   def send_notification(customer_id):
       # 调用SES发送邮件，或通过WebSocket API推送消息
       # 示例：SES发送邮件
       ses = boto3.client('ses')
       ses.send_email(
           Source='your-notification@example.com',
           Destination={'ToAddresses': [f'{customer_id}@example.com']},
           Message={
               'Subject': {'Data': '消息处理完成通知'},
               'Body': {'Text': {'Data': '您的1000条消息已全部处理完毕'}}
           }
       )
   

3. 处理幂等与重试

   • 给每条SQS消息添加唯一message_id，Lambda处理前先查询DynamoDB是否已处理该ID，避免重复计数。
   • 利用DynamoDB的条件更新，确保只有当notified为false时才触发通知，防止重复发送。

方案二：FIFO队列分组+CloudWatch告警（适合轻量场景） #

如果使用SQS FIFO队列，可以给客户A的所有消息设置相同的MessageGroupId: "customer-A"，通过CloudWatch跟踪消息处理量来触发通知。

具体步骤 #

1. 配置FIFO队列与消息分组
   创建SQS FIFO队列，推送客户A的消息时指定MessageGroupId为customer-A，确保同组消息按顺序处理（可选，不影响计数）。

2. 创建CloudWatch指标与告警

   • 跟踪SQS的NumberOfMessagesDeleted指标，过滤维度为QueueName和MessageGroupId=customer-A。
   • 当该指标的累计值达到1000时，触发CloudWatch告警，关联SNS主题来发送邮件通知，或调用Lambda推送WebSocket消息。

3. 注意事项

   • 该方案无法区分消息是成功处理还是进入死信队列，需配合死信队列的监控，确保1000条是成功处理的消息。
   • 若存在消息重试，可能导致计数虚高，需结合Lambda的SuccessInvocations指标（过滤客户A）来统计更准确。

方案三：批量处理+全局计数器（适合高吞吐量场景） #

如果Lambda配置了批量消费SQS消息，可以批量统计客户A的消息数量，更新全局计数器，达到阈值时触发通知。

具体步骤 #

1. Lambda批量消费配置
   在Lambda触发器设置中，将SQS批量大小设为合适值（比如100），减少DynamoDB调用次数。

2. 批量更新计数器
   Lambda每次处理批量消息时，统计其中客户A的消息数量，一次性更新DynamoDB的processed_count：

   def lambda_handler(event, context):
       customer_a_count = sum(1 for record in event['Records'] if record['body'].get('customer_id') == 'A')
       if customer_a_count == 0:
           return
   
       response = table.update_item(
           Key={'customer_id': 'A'},
           UpdateExpression='ADD processed_count :inc',
           ExpressionAttributeValues={':inc': customer_a_count},
           ReturnValues='ALL_NEW'
       )
       new_item = response['Attributes']
   
       if new_item['processed_count'] == new_item['total_messages'] and not new_item['notified']:
           table.update_item(
               Key={'customer_id': 'A'},
               UpdateExpression='SET notified = :val',
               ExpressionAttributeValues={':val': True}
           )
           send_notification('A')
   

3. 优势
   减少DynamoDB的调用次数，提升处理效率，适合高吞吐量的场景。

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