Cloud Tasks 条件执行

Question

我正在使用 Cloud Tasks。只有在任务 A 和任务 B 成功完成后，我才需要触发任务 C 的执行。所以我需要某种方式来阅读/被通知触发的任务状态。但是我在 GCP 的文档中看不到这样做的方法。如果有帮助的话，使用 Node.js SDK 创建任务和 Cloud Functions 作为任务处理程序。

编辑：

根据要求，这里有更多关于我们正在做的事情的信息：

任务 1 到 10 分别发出 HTTP 请求、获取数据、根据这些数据更新 Firestore 中的各个集合。这 10 个任务可以并行运行，并且没有特定的顺序，因为它们彼此之间没有任何依赖关系。所有这些任务实际上都是在 GCF 内部实现的。

Task 11实际上依赖于Tasks 1-10更新的Firestore集合数据，所以只有Tasks 1-10成功完成后才能运行。

我们确实会发出一个 RunID 作为通用标识符来对所有任务 (1 - 11) 的特定运行进行分组。

Answer 1

Cloud Task 只触发任务，只能定义时间条件。您必须在任务 C 运行时手动编写检查代码。

这里是一个流程示例：

• 任务 A 正在运行，最后，写入 firestore 的任务已完成
• 任务 B 正在运行，最后，写入 firestore 的任务已完成
• 任务 C 启动并检查 A 和 B 是否在 firestore 中完成。
  • 如果不是，任务退出错误
  • 是，继续这个过程

您必须自定义 C 任务队列，以便在出错时重试任务。

另一个昂贵的解决方案是使用 Cloud Composer 来处理此工作流

目前没有其他关于工作流管理的解决方案。

Answer 2

Cloud Tasks 不是您想在这种情况下使用的工具。查看Cloud Composer，它内置于 Apache Airflow for GCP 之上。

编辑：您可以创建一个 GCF 来处理这些请求的状态

import requests
from concurrent.futures import ThreadPoolExecutor, as_completed

################ TASK A
taskA_list = [
    "https://via.placeholder.com/400",
    "https://via.placeholder.com/410",
    "https://via.placeholder.com/420",
    "https://via.placeholder.com/430",
    "https://via.placeholder.com/440",
    "https://via.placeholder.com/450",
    "https://via.placeholder.com/460",
    "https://via.placeholder.com/470",
    "https://via.placeholder.com/480",
    "https://via.placeholder.com/490",
]

def call2TaskA(url):
    html = requests.get(url, stream=True)
    return (url,html.status_code)


processes = []
results = []
with ThreadPoolExecutor(max_workers=10) as executor:
    for url in taskA_list:
        processes.append(executor.submit(call2TaskA, url))

isOkayToDoTaskB = True
for taskA in as_completed(processes):
    result = taskA.result()
    if result[1] != 200: # your validation on taskA
        isOkayToDoTaskB = False
    results.append(result)

if not isOkayToDoTaskB:
    raise ValueError('Problems: {}'.format(results))

################ TASK B
def doTaskB():
    pass

doTaskB()