如何结合EventBridge两类事件，实现S3对象双条件触发任务？

核心思路 #

通过状态跟踪+条件触发的方式，用持久化存储记录每个Ax的上传状态、当前B的有效性以及任务执行状态，确保仅当两个依赖条件都满足时才触发任务，且每个Ax仅对应一次任务执行。

具体实现步骤 #

1. 搭建状态存储（DynamoDB） #

创建一张DynamoDB表，包含两种核心条目类型：

• Ax任务状态项：
  • 主键：ItemKey（值为S3中Ax的对象键，例如A1/dataset/file.csv）
  • 属性：
    • AxUploaded：布尔值，标记该Ax是否已完成上传
    • CurrentBValid：布尔值，标记当前周期的B是否处于有效状态
    • TaskExecuted：布尔值，标记该Ax对应的任务是否已执行
    • BVersion：字符串，记录当前有效B的版本标识（如生成时间戳、批次ID）
• B全局状态项：
  • 主键：ItemKey（固定值如B_GLOBAL_STATUS）
  • 属性：
    • IsValid：布尔值，标记当前B是否处于有效周期
    • Version：字符串，当前有效B的版本标识

2. 处理对象B的周期生成与状态同步 #

• 用EventBridge Scheduler按指定周期触发B的生成流程，生成完成后执行：
  1. 更新B全局状态项：将IsValid设为true，Version设为本次生成的版本号
  2. 批量更新DynamoDB中所有TaskExecuted=false的Ax任务状态项：同步CurrentBValid为true、BVersion为新版本号
  3. 遍历这些未执行的Ax项，对AxUploaded=true的条目，直接触发对应任务，并通过DynamoDB条件更新将TaskExecuted设为true（保证原子性）

3. 处理Ax的上传事件 #

当S3完成Ax上传时，通过EventBridge触发Lambda函数，执行以下逻辑：

1. 初始化/更新Ax状态：
   • 按Ax的对象键查询DynamoDB，若不存在则创建条目，默认AxUploaded=true，CurrentBValid从B全局状态项获取，TaskExecuted=false
   • 若条目已存在，直接更新AxUploaded=true
2. 条件触发任务：
   • 检查当前CurrentBValid=true且TaskExecuted=false
   • 若满足条件，调用DynamoDB条件更新（ConditionExpression: TaskExecuted = false）将TaskExecuted设为true，同时启动对应的任务实例

4. 关键保障机制 #

• 原子性控制：所有状态更新和任务触发的判断，必须通过DynamoDB条件表达式实现，避免并发场景下的重复执行
• B失效处理：当B进入新周期需重新生成时，先将B全局状态项的IsValid设为false，此时新上传的Ax不会触发任务，直到B生成完成后再同步状态
• 历史清理：定期清理DynamoDB中TaskExecuted=true的旧条目，避免存储冗余

内容的提问来源于stack exchange，提问作者f.khantsis