使用Lambda追加更新S3文件的并发问题及优化方案咨询

已知通过Lambda更新同一S3文件并非最佳实践，但业务要求必须在指定S3文件中写入并追加计数结果。正常情况下作业会按顺序逐个更新计数，但繁忙时段Lambda定时作业可能并发执行，导致文件更新异常甚至损坏。考虑过将Lambda并发数设为0强制串行执行，但担心引发性能问题，现寻求更优的并发防护方案或代码错误处理优化建议。

现有代码 #

lambda_handler.py #

import json
from s3 import *

def lambda_handler(event, context):
    s3_append('yyyymmdd/result.txt', 'yyyymmdd/result.txt', 'append str')

S3.py #

import boto3
from datetime import datetime

bucket_name = 's3-sachiko-aws3-new2'
s3_resource = boto3.resource("s3")

# S3 File Append
def s3_append(target_path, append_path, append_str):
    target_context = s3_resource.Object(bucket_name, target_path).get()["Body"].read()  
    s3_resource.Object(bucket_name, append_path).put(Body=target_context + b"\n" +  bytes(append_str, 'utf-8'))

1. S3乐观锁（条件写入） #

利用S3的ETag实现乐观锁，避免并发覆盖。核心逻辑是：读取文件时记录ETag，写入时仅当文件当前ETag与记录值一致才执行操作，不匹配则重试。

修改后的S3.py示例：

import boto3
import botocore
import time

bucket_name = 's3-sachiko-aws3-new2'
s3_resource = boto3.resource("s3")

# S3 File Append with optimistic lock
def s3_append(target_path, append_path, append_str, max_retries=3):
    retries = 0
    while retries < max_retries:
        try:
            # 获取文件内容及ETag
            obj = s3_resource.Object(bucket_name, target_path)
            response = obj.get()
            target_context = response["Body"].read()
            etag = response["ETag"].strip('"')  # 去除ETag的引号
            
            # 条件写入，仅当ETag匹配时更新
            obj.put(
                Body=target_context + b"\n" + bytes(append_str, 'utf-8'),
                IfMatch=etag
            )
            return
        except botocore.exceptions.ClientError as e:
            if e.response['Error']['Code'] == 'PreconditionFailed':
                # ETag不匹配，重试
                retries += 1
                time.sleep(0.5)
            else:
                # 其他异常抛出
                raise
        except s3_resource.meta.client.exceptions.NoSuchKey:
            # 文件不存在，直接写入初始内容
            s3_resource.Object(bucket_name, append_path).put(Body=bytes(append_str, 'utf-8'))
            return
    # 重试超过上限，抛出异常
    raise Exception(f"Failed to append to S3 file after {max_retries} retries")

2. 分布式锁（基于DynamoDB） #

通过DynamoDB实现分布式锁，确保同一时间只有一个Lambda实例能操作目标文件：

• 创建DynamoDB表，主键设为lock_key（对应S3文件路径），额外添加expire_time字段（防止Lambda崩溃导致锁长期占用）
• Lambda执行前，用PutItem的条件表达式判断锁是否不存在或已过期，以此获取锁
• 操作完成后删除锁；若获取锁失败，等待后重试或根据业务需求返回

3. 队列缓冲+批量写入 #

如果计数操作频率高，可引入SQS队列缓冲请求：

• 将每次计数请求发送到SQS队列
• 配置单并发Lambda（或固定实例ECS任务）消费队列，批量收集计数结果后一次性写入S3
• 既避免了并发写入冲突，又通过批量处理提升写入效率

4. 代码基础优化 #

• 异常补全：捕获文件不存在、S3连接超时等场景，避免程序无预期崩溃
• 日志记录：添加关键步骤日志（如ETag获取、写入结果、重试次数），方便排查问题
• 参数校验：对target_path、append_str等参数做合法性校验，避免无效请求

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