如何跟踪Google App Engine中已触发Cloud Task队列的全部执行完成状态？

针对你两级嵌套触发push队列的场景，下面是可直接落地的实现方案：

核心实现思路 #

你需要通过外置持久化存储统一维护任务组的执行状态，所有任务执行完成后主动触发下一阶段逻辑，无需依赖平台控制台查询。

具体实现步骤 #

方案1：计数统计法（推荐，性能最高） #

该方案通过计数器实时比对任务完成进度，实现成本最低：

• 提前准备持久化存储（Redis、云数据库、KV存储均可），为本次触发的所有任务生成唯一group_id作为标识，初始化状态记录：
  记录字段包含：总任务数、已完成任务数、第一级任务完成数、可选的失败任务列表
• 触发第一级循环前，先把第一级任务数len(listA)计入总任务数
• 给两级任务的处理逻辑各加状态上报逻辑：

  # taskqueue1 执行逻辑
  def run_taskqueue1(payload):
      group_id = payload["group_id"]
      listB = get_related_listB(payload["item"])
      # 触发第二级任务
      second_task_num = len(listB)
      for b in listB:
          taskqueue2.push({"group_id": group_id, "data": b})
      # 上报状态：总任务数加上当前触发的第二级任务数，第一级完成数+1，已完成数+1（taskqueue1本身执行完成）
      update_state(group_id, inc_total=second_task_num, inc_first_finish=1, inc_finish=1)
      # 检查是否全部完成
      check_is_all_done(group_id)
  
  # taskqueue2 执行逻辑
  def run_taskqueue2(payload):
      group_id = payload["group_id"]
      # 执行业务逻辑
      do_your_business(payload["data"])
      # 上报状态：已完成数+1
      update_state(group_id, inc_finish=1)
      # 检查是否全部完成
      check_is_all_done(group_id)
  

• 完成判断逻辑：

  def check_is_all_done(group_id):
      state = get_state(group_id)
      # 第一级任务全部执行完成后，总任务数才完全确定，此时再判断完成状态
      if state.first_level_finish == len(listA) and state.finish_task == state.total_task:
          # 所有任务全部完成，进入下一阶段
          run_next_step()
  

方案2：痕迹留档法（适合需要排查问题的场景） #

如果需要留存每个任务的执行记录，可采用标记法：

• 给所有任务加同一个group_id标识，每个任务执行完成后，在存储里插入一条唯一的完成记录，包含任务ID、执行状态、错误信息等字段
• 需要判断完成状态时，直接按group_id统计所有记录的数量、状态，和总任务数比对即可

注意事项 #

• 要做任务去重：同一个任务多次重试时，仅上报一次完成/失败状态，避免重复计数导致统计错误，可以通过任务唯一ID做幂等判断
• 异常兜底：可以额外设置超时时间、失败任务告警逻辑，避免任务永久失败导致流程卡住
• 可靠性保障：状态上报逻辑可以和任务业务逻辑放在同一个事务里执行，避免业务执行成功但状态上报失败的不一致问题

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