如何实现Lambda与S3 Bucket同步，避免文件重复处理？

针对你的Lambda函数因S3事件和调度器重复触发导致文件多次处理的问题，以下是几个可靠的同步实现方案，核心都是通过分布式锁或原子操作实现"发现→锁定→处理→解锁"的流程：

方案一：基于DynamoDB的分布式锁（推荐，可靠性高） #

利用DynamoDB的条件写入特性实现原子锁，避免竞态条件，适合需要严格锁控制的场景。

步骤1：创建DynamoDB锁表 #

创建一张DynamoDB表，配置如下：

• 主键：file_identifier（字符串类型，用{源桶名}/{文件键}作为唯一标识）
• 可选字段：
  • lock_expiry：数值类型，存储锁的过期时间戳（Unix秒级时间，防止Lambda异常挂起导致锁永久占用）
  • lambda_request_id：字符串类型，记录持有锁的Lambda请求ID，用于精准释放锁

步骤2：Lambda锁逻辑实现 #

import boto3
import time

dynamodb = boto3.resource('dynamodb')
lock_table = dynamodb.Table('s3-file-processing-locks')
s3 = boto3.client('s3')

def get_file_identifier(bucket, key):
    return f"{bucket}/{key}"

def acquire_lock(file_id, context, lock_duration=1800):
    """获取锁，默认锁定30分钟（需长于你的最大文件处理时间）"""
    expiry_time = int(time.time()) + lock_duration
    try:
        lock_table.put_item(
            Item={
                'file_identifier': file_id,
                'lock_expiry': expiry_time,
                'lambda_request_id': context.aws_request_id
            },
            ConditionExpression="attribute_not_exists(file_identifier) OR lock_expiry < :now",
            ExpressionAttributeValues={':now': int(time.time())}
        )
        return True
    except dynamodb.meta.client.exceptions.ConditionalCheckFailedException:
        # 锁已存在且未过期，获取失败
        return False

def release_lock(file_id, context):
    """释放锁，仅删除当前Lambda持有的锁"""
    try:
        lock_table.delete_item(
            Key={'file_identifier': file_id},
            ConditionExpression="lambda_request_id = :req_id",
            ExpressionAttributeValues={':req_id': context.aws_request_id}
        )
    except Exception:
        # 锁已过期自动清理或其他异常，无需处理
        pass

def process_file(source_bucket, source_key, target_bucket):
    """替换为你的实际文件处理逻辑"""
    # 示例：复制文件到目标桶后删除源文件
    s3.copy_object(
        Bucket=target_bucket,
        Key=source_key,
        CopySource={'Bucket': source_bucket, 'Key': source_key}
    )
    s3.delete_object(Bucket=source_bucket, Key=source_key)

def lambda_handler(event, context):
    target_bucket = "your-target-bucket"
    # 处理S3文件创建事件
    if 'Records' in event:
        for record in event['Records']:
            source_bucket = record['s3']['bucket']['name']
            source_key = record['s3']['object']['key']
            file_id = get_file_identifier(source_bucket, source_key)
            
            if acquire_lock(file_id, context):
                try:
                    process_file(source_bucket, source_key, target_bucket)
                finally:
                    release_lock(file_id, context)
    # 处理调度器触发的批量扫描
    else:
        source_bucket = "your-source-bucket"
        paginator = s3.get_paginator('list_objects_v2')
        for page in paginator.paginate(Bucket=source_bucket):
            if 'Contents' not in page:
                continue
            for obj in page['Contents']:
                source_key = obj['Key']
                file_id = get_file_identifier(source_bucket, source_key)
                
                if acquire_lock(file_id, context):
                    try:
                        process_file(source_bucket, source_key, target_bucket)
                    finally:
                        release_lock(file_id, context)

方案二：纯S3原子操作实现锁（无额外服务，轻量） #

利用S3的原子复制/移动操作，将待处理文件临时转移到processing/前缀下，从根源避免重复扫描和处理。

核心流程 #

1. 发现文件后，尝试将其原子移动到源桶/processing/目录（复制+删除原文件）
2. 移动成功则视为锁定，开始处理文件
3. 处理完成后将文件移到目标桶，失败则移回原目录

Lambda代码示例 #

import boto3
import urllib.parse

s3 = boto3.client('s3')

def lock_file(source_bucket, source_key):
    processing_key = f"processing/{source_key}"
    try:
        # 原子复制：仅当原文件存在且processing目录无该文件时执行
        s3.copy_object(
            Bucket=source_bucket,
            Key=processing_key,
            CopySource={'Bucket': source_bucket, 'Key': source_key},
            ConditionExpression="exists(CopySource) AND not exists(Key)"
        )
        # 复制成功后删除原文件，完成锁定
        s3.delete_object(Bucket=source_bucket, Key=source_key)
        return processing_key
    except s3.exceptions.ConditionalCheckFailedException:
        # 文件已被锁定或不存在
        return None

def process_and_unlock(source_bucket, processing_key, target_bucket):
    """处理文件并解锁（移到目标桶）"""
    # 替换为你的实际文件处理逻辑
    response = s3.get_object(Bucket=source_bucket, Key=processing_key)
    content = response['Body'].read().decode('utf-8')
    
    # 处理完成后移到目标桶
    target_key = processing_key.replace('processing/', '')
    s3.copy_object(
        Bucket=target_bucket,
        Key=target_key,
        CopySource={'Bucket': source_bucket, 'Key': processing_key}
    )
    s3.delete_object(Bucket=source_bucket, Key=processing_key)

def lambda_handler(event, context):
    target_bucket = "your-target-bucket"
    source_bucket = "your-source-bucket"
    
    if 'Records' in event:
        for record in event['Records']:
            source_key = urllib.parse.unquote_plus(record['s3']['object']['key'], encoding='utf-8')
            processing_key = lock_file(source_bucket, source_key)
            if processing_key:
                try:
                    process_and_unlock(source_bucket, processing_key, target_bucket)
                except Exception as e:
                    # 处理失败，将文件移回原目录
                    original_key = processing_key.replace('processing/', '')
                    s3.copy_object(
                        Bucket=source_bucket,
                        Key=original_key,
                        CopySource={'Bucket': source_bucket, 'Key': processing_key}
                    )
                    s3.delete_object(Bucket=source_bucket, Key=processing_key)
                    raise e
    else:
        # 调度器触发，仅扫描源桶根目录（跳过processing前缀）
        paginator = s3.get_paginator('list_objects_v2')
        for page in paginator.paginate(
            Bucket=source_bucket,
            Prefix='',
            Delimiter='/'
        ):
            if 'Contents' not in page:
                continue
            for obj in page['Contents']:
                source_key = obj['Key']
                if source_key.startswith('processing/'):
                    continue
                processing_key = lock_file(source_bucket, source_key)
                if processing_key:
                    try:
                        process_and_unlock(source_bucket, processing_key, target_bucket)
                    except Exception as e:
                        original_key = processing_key.replace('processing/', '')
                        s3.copy_object(
                            Bucket=source_bucket,
                            Key=original_key,
                            CopySource={'Bucket': source_bucket, 'Key': processing_key}
                        )
                        s3.delete_object(Bucket=source_bucket, Key=processing_key)
                        raise e

方案三：S3对象标签锁（轻量但需注意一致性） #

通过给文件添加processing=true标签标记锁定状态，处理完成后移除标签，适合对锁精度要求不高的场景。

核心代码片段 #

def acquire_lock_via_tag(bucket, key):
    try:
        s3.put_object_tagging(
            Bucket=bucket,
            Key=key,
            Tagging={'TagSet': [{'Key': 'processing', 'Value': 'true'}]},
            ConditionExpression="not exists(tagging['processing']) OR tagging['processing'] = :false",
            ExpressionAttributeValues={':false': 'false'}
        )
        return True
    except s3.exceptions.ConditionalCheckFailedException:
        return False

def release_lock_via_tag(bucket, key):
    # 移除processing标签
    existing_tags = s3.get_object_tagging(Bucket=bucket, Key=key)['TagSet']
    updated_tags = [tag for tag in existing_tags if tag['Key'] != 'processing']
    s3.put_object_tagging(
        Bucket=bucket,
        Key=key,
        Tagging={'TagSet': updated_tags}
    )

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