基于AWS Lambda批量转存SQS消息至S3的代码及指引请求

核心思路 #

通过Lambda监听SQS队列，每次拉取批量消息后累积存储，当累积数量达到10万条时，将消息序列化为JSON格式写入S3；用DynamoDB保存当前累积的消息和计数，解决Lambda跨执行上下文的状态保持问题，同时通过乐观锁避免并发执行时的计数冲突。

Lambda示例代码（Python） #

import boto3
import json
from datetime import datetime

# 初始化客户端
s3 = boto3.client('s3')
dynamodb = boto3.client('dynamodb')
sqs = boto3.client('sqs')

# 配置参数（根据实际环境修改）
S3_BUCKET = 'your-target-archive-bucket'
DYNAMO_TABLE = 'sqs-batch-tracker'
BATCH_SIZE_THRESHOLD = 100000
BATCH_ID = 'daily-sqs-archive'
SQS_QUEUE_URL = 'your-sqs-queue-url'

def lambda_handler(event, context):
    # 提取SQS消息内容和回执句柄
    new_messages = []
    receipt_handles = []
    for record in event['Records']:
        new_messages.append(json.loads(record['body']))
        receipt_handles.append(record['receiptHandle'])
    
    # 从DynamoDB获取当前累积批次数据
    try:
        db_response = dynamodb.get_item(
            TableName=DYNAMO_TABLE,
            Key={'batch_id': {'S': BATCH_ID}}
        )
        current_batch = db_response.get('Item', {})
        accumulated_msgs = json.loads(current_batch.get('messages', {'S': '[]'})['S'])
        current_count = int(current_batch.get('current_count', {'N': '0'})['N'])
    except Exception as e:
        print(f"读取批次数据失败: {str(e)}")
        accumulated_msgs = []
        current_count = 0
    
    # 合并新消息并更新计数
    accumulated_msgs.extend(new_messages)
    updated_count = current_count + len(new_messages)
    
    # 检查是否达到批量写入阈值
    if updated_count >= BATCH_SIZE_THRESHOLD:
        # 生成唯一S3文件名
        timestamp = datetime.utcnow().strftime('%Y%m%d-%H%M%S-%f')
        s3_key = f'archive/{timestamp}-batch-{BATCH_ID}.json'
        
        # 写入S3
        try:
            s3.put_object(
                Bucket=S3_BUCKET,
                Key=s3_key,
                Body=json.dumps(accumulated_msgs),
                ContentType='application/json'
            )
            print(f"成功写入{updated_count}条消息到S3: {s3_key}")
            
            # 重置DynamoDB批次数据
            dynamodb.put_item(
                TableName=DYNAMO_TABLE,
                Item={
                    'batch_id': {'S': BATCH_ID},
                    'messages': {'S': '[]'},
                    'current_count': {'N': '0'}
                }
            )
            
            # 删除已处理的SQS消息
            batch_delete_sqs_messages(receipt_handles)
        except Exception as e:
            print(f"S3写入失败: {str(e)}")
            return {'statusCode': 500, 'body': '批量写入S3失败'}
    else:
        # 更新DynamoDB累积数据（乐观锁避免并发冲突）
        try:
            dynamodb.put_item(
                TableName=DYNAMO_TABLE,
                Item={
                    'batch_id': {'S': BATCH_ID},
                    'messages': {'S': json.dumps(accumulated_msgs)},
                    'current_count': {'N': str(updated_count)}
                },
                ConditionExpression='current_count = :val',
                ExpressionAttributeValues={':val': {'N': str(current_count)}}
            )
            
            # 删除已处理的SQS消息
            batch_delete_sqs_messages(receipt_handles)
        except dynamodb.exceptions.ConditionalCheckFailedException:
            print("检测到并发更新冲突，将重试该批次")
            return {'statusCode': 409, 'body': '并发更新冲突'}
        except Exception as e:
            print(f"更新批次数据失败: {str(e)}")
            return {'statusCode': 500, 'body': '批次数据更新失败'}
    
    return {'statusCode': 200, 'body': f"处理了{len(new_messages)}条消息"}

def batch_delete_sqs_messages(receipt_handles):
    """批量删除SQS消息（单次最多删10条）"""
    for i in range(0, len(receipt_handles), 10):
        batch = receipt_handles[i:i+10]
        entries = [{'Id': str(idx), 'ReceiptHandle': handle} for idx, handle in enumerate(batch)]
        try:
            sqs.delete_message_batch(QueueUrl=SQS_QUEUE_URL, Entries=entries)
        except Exception as e:
            print(f"删除消息批次失败: {str(e)}")

配置与部署指引 #

1. 前置资源准备 #

• S3存储桶：创建归档用的S3桶，可选开启版本控制防止数据丢失。
• DynamoDB表：创建表，主键为batch_id（字符串类型），无需额外索引。
• SQS队列：设置可见性超时大于Lambda超时时间（建议10分钟），避免消息重复处理。

2. Lambda配置 #

• 权限：给Lambda角色添加以下权限：
  • SQS：sqs:ReceiveMessage、sqs:DeleteMessageBatch
  • S3：s3:PutObject
  • DynamoDB：dynamodb:GetItem、dynamodb:PutItem
• 触发配置：添加SQS事件源，设置Batch size为1000（SQS给Lambda的最大批量值），Batch window设为5分钟（可选，用于累积更多消息再触发）。
• 运行配置：内存设为512MB，超时设为5分钟，确保有足够资源处理批量消息。

3. 异常处理建议 #

• 消息重复：如果Lambda执行失败，SQS会重发消息，建议在业务层添加唯一ID做幂等校验。
• 超时处理：可在代码中添加Lambda超时检测，若接近超时且累积消息较多，提前写入S3避免数据丢失。
• 并发冲突：代码中已加入乐观锁，冲突时Lambda会自动重试该批次消息。

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