如何监听AWS S3 Bucket文件上传并在网站中处理文件？

下面是几种成熟的实现方案，根据你的需求场景选择：

1. AWS Lambda + S3事件通知（推荐） #

这是AWS生态下最常用、成本最低的托管式方案，完全不需要维护服务器。

实现步骤： #

• 1.1 在AWS控制台创建Lambda函数，编写处理逻辑（示例Python代码如下）：

  import boto3
  import requests
  
  def lambda_handler(event, context):
      s3_client = boto3.client('s3')
      # 遍历S3事件中的上传记录
      for record in event['Records']:
          bucket = record['s3']['bucket']['name']
          file_key = record['s3']['object']['key']
          
          # 从S3获取文件内容
          s3_response = s3_client.get_object(Bucket=bucket, Key=file_key)
          file_content = s3_response['Body'].read()
          
          # 调用你的网站处理流程（示例：POST到网站API）
          requests.post(
              "https://your-site.com/api/process-file",
              files={'uploaded_file': (file_key, file_content)}
          )
      
      return {'statusCode': 200, 'body': '文件处理触发成功'}
  

• 1.2 配置S3 Bucket的事件通知：
  • 进入目标Bucket的“属性”→“事件通知”，创建新通知
  • 事件类型选择所有对象创建事件（或按需选“Put”“Post”等特定事件）
  • 目标选择“Lambda函数”，指定你创建的Lambda函数
• 1.3 配置权限：
  • 确保Lambda的执行角色拥有s3:GetObject权限（访问目标Bucket）
  • S3需要拥有触发Lambda的权限（配置事件通知时可自动生成对应的IAM策略）

优势： #

• 按需付费，无请求时几乎零成本
• 毫秒级触发延迟
• AWS自动处理扩缩容、故障重试

2. S3 + Event Bridge（灵活集成方案） #

如果需要更复杂的事件路由、过滤，或者要集成Lambda之外的服务（比如ECS、自建服务器），可以用Event Bridge中转事件。

实现步骤： #

• 2.1 配置S3事件通知，目标选择“EventBridge”
• 2.2 进入Event Bridge控制台，创建规则：
  • 事件源选择“AWS事件”，匹配模式设置为：

    source: aws.s3
    detail-type: Object Created
    detail.bucket.name: your-bucket-name
    

  • 按需添加过滤条件（比如只处理.csv后缀文件：detail.object.key:Suffix: .csv）
  • 目标选择你的处理服务：比如ECS任务、HTTP API端点（Event Bridge支持直接POST到第三方API）

优势： #

• 支持更精细的事件过滤
• 可对接更多AWS服务或自建服务
• 事件可留存、重放

3. 自建轮询脚本（非托管方案） #

如果不想依赖AWS托管服务，可自行编写定时轮询脚本，定期检查Bucket中的新增文件。

实现思路： #

• 编写脚本（示例Python代码），定期调用S3的list_objects_v2接口，对比上次检查时间，筛选出新增文件：

  import boto3
  from datetime import datetime, timedelta
  import os
  
  def poll_s3_for_new_files():
      s3_client = boto3.client('s3')
      bucket_name = "your-bucket"
      last_check_path = "last_check_timestamp.txt"
      
      # 读取上次检查时间（首次运行设为1小时前）
      if os.path.exists(last_check_path):
          with open(last_check_path, 'r') as f:
              last_check_time = datetime.fromisoformat(f.read().strip())
      else:
          last_check_time = datetime.now() - timedelta(hours=1)
      
      # 获取Bucket中所有文件
      paginator = s3_client.get_paginator('list_objects_v2')
      for page in paginator.paginate(Bucket=bucket_name):
          for obj in page.get('Contents', []):
              if obj['LastModified'].replace(tzinfo=None) > last_check_time:
                  # 获取文件并调用处理流程
                  file_content = s3_client.get_object(Bucket=bucket_name, Key=obj['Key'])['Body'].read()
                  # 替换为你的网站处理逻辑
                  # process_file(obj['Key'], file_content)
      
      # 更新上次检查时间
      with open(last_check_path, 'w') as f:
          f.write(datetime.now().isoformat())
  
  if __name__ == "__main__":
      poll_s3_for_new_files()
  

• 用定时任务工具（比如Linux cron、Windows任务计划）定期执行脚本（比如每5分钟一次）

注意点： #

• 存在检查间隔的延迟（比如5分钟间隔，最多延迟5分钟）
• 需要自行处理分页、重复文件、API调用限额等问题
• 长期运行需要维护脚本和服务器

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