使用 Cloud Tasks 队列缓冲工作流执行

本教程介绍如何创建可调节工作流执行速率的 Cloud Tasks 队列。

可以同时进行的工作流执行次数存在上限。一旦此配额用尽，并且如果执行积压处于停用状态，或者如果积压的执行作业达到配额上限，则任何新的执行作业都会失败，并返回 HTTP 429 Too many requests 状态代码。通过使 Cloud Tasks 队列能够以您定义的速率执行子工作流，您可以避免与 Workflows 配额相关的问题，并实现更高的执行速率。

请注意，Cloud Tasks 旨在提供“至少一次”提交；不过，对于来自 Cloud Tasks 的重复请求，Workflows 无法确保只处理一次。

在下图中，父工作流会调用受应用了调度率的 Cloud Tasks 队列监管的子工作流。

目标

在此教程中，您将学习以下操作：

创建充当父工作流和子工作流之间中介的 Cloud Tasks 队列。
创建并部署可从父工作流接收数据的子工作流。
创建并部署通过 Cloud Tasks 队列执行子工作流的父工作流。
运行没有调度速率限制的父工作流，该工作流会调用子工作流的执行。
对 Cloud Tasks 队列应用调度限制，然后运行父工作流。
请注意，子工作流的执行速率由 Cloud Tasks 队列定义。

您可以在 Google Cloud 控制台中运行以下命令，也可以在终端或 Cloud Shell 中使用 Google Cloud CLI 运行这些命令。

费用

在本文档中，您将使用 Google Cloud的以下收费组件：

如需根据您的预计使用情况来估算费用，请使用价格计算器。

新 Google Cloud 用户可能有资格申请免费试用。

准备工作

您的组织定义的安全限制条件可能会导致您无法完成以下步骤。如需了解相关问题排查信息，请参阅在受限的 Google Cloud 环境中开发应用。

控制台

登录您的 Google 账号。

如果您还没有 Google 账号，请注册新账号。

In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

Roles required to select or create a project

Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

Go to project selector

Verify that billing is enabled for your Google Cloud project.

Enable the Cloud Tasks, Compute Engine, and Workflows APIs.

Roles required to enable APIs

To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

Enable the APIs

In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

Roles required to select or create a project

Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

Go to project selector

Verify that billing is enabled for your Google Cloud project.

Enable the Cloud Tasks, Compute Engine, and Workflows APIs.

Roles required to enable APIs

To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

Enable the APIs

在 Google Cloud 控制台中，前往 IAM 页面，为 Compute Engine 默认服务账号设置权限。
转到 IAM

记下 Compute Engine 默认服务账号，因为您将在本教程中将其与工作流关联以进行测试。启用或使用包含 Compute Engine 的 Google Cloud 服务后，系统会自动创建此服务账号，其电子邮件地址格式如下：
```
PROJECT_NUMBER-compute@developer.gserviceaccount.com
```
将 PROJECT_NUMBER 替换为您的Google Cloud 项目编号。您可以在 Google Cloud 控制台的欢迎页面上找到项目编号。

对于生产环境，我们强烈建议创建新的服务账号，并为其授予一个或多个 IAM 角色，这些角色包含所需的最小权限并遵循最小权限原则。

注意：
根据您的组织政策配置，默认服务账号可能会自动获得项目的 Editor 角色。我们强烈建议您通过强制执行 iam.automaticIamGrantsForDefaultServiceAccounts 组织政策限制条件来停用自动角色授予功能。如果您的组织是在 2024 年 5 月 3 日之后创建的，则默认情况下会强制执行此限制条件。

如果您停用自动角色授予功能，则必须决定向默认服务账号授予哪些角色，然后自行授予这些角色。

如果默认服务账号已具有 Editor 角色，我们建议您将 Editor 角色替换为权限较少的角色。如需安全地修改服务账号的角色，请使用 Policy Simulator 查看更改的影响，然后授予和撤消相应的角色。
选择 Compute Engine 默认服务账号，然后在相应行中点击 修改主账号。
在随即显示的对话框中，点击 add 添加其他角色，然后添加以下角色：
1. 在选择角色列表中，选择 Workflows > Workflows Invoker，以便该账号有权触发您的工作流执行。
2. 在选择角色列表中，选择 Cloud Tasks > Cloud Tasks Enqueuer，以便账号拥有创建任务的权限。
点击保存。

gcloud

登录您的 Google 账号。

如果您还没有 Google 账号，请注册新账号。

安装 Google Cloud CLI。

如果您使用的是外部身份提供方 (IdP)，则必须先使用联合身份登录 gcloud CLI。

如需初始化 gcloud CLI，请运行以下命令：

gcloud init

创建或选择 Google Cloud 项目。

选择或创建项目所需的角色

选择项目：选择项目不需要特定的 IAM 角色，您可以选择已获授角色的任何项目。
创建项目：如需创建项目，您需要拥有 Project Creator 角色 (roles/resourcemanager.projectCreator)，该角色包含 resourcemanager.projects.create 权限。了解如何授予角色。

创建 Google Cloud 项目：
```
gcloud projects create PROJECT_ID
```
将 PROJECT_ID 替换为您要创建的 Google Cloud 项目的名称。
选择您创建的 Google Cloud 项目：
```
gcloud config set project PROJECT_ID
```
将 PROJECT_ID 替换为您的 Google Cloud 项目名称。

验证是否已为您的 Google Cloud 项目启用结算功能。

启用 Cloud Tasks、Compute Engine 和 Workflows API：

启用 API 所需的角色

如需启用 API，您需要拥有 Service Usage Admin IAM 角色 (roles/serviceusage.serviceUsageAdmin)，该角色包含 serviceusage.services.enable 权限。了解如何授予角色。

gcloud services enable cloudtasks.googleapis.com compute.googleapis.com workflows.googleapis.com

安装 Google Cloud CLI。

如果您使用的是外部身份提供方 (IdP)，则必须先使用联合身份登录 gcloud CLI。

如需初始化 gcloud CLI，请运行以下命令：

gcloud init

创建或选择 Google Cloud 项目。

选择或创建项目所需的角色

选择项目：选择项目不需要特定的 IAM 角色，您可以选择已获授角色的任何项目。
创建项目：如需创建项目，您需要拥有 Project Creator 角色 (roles/resourcemanager.projectCreator)，该角色包含 resourcemanager.projects.create 权限。了解如何授予角色。

创建 Google Cloud 项目：
```
gcloud projects create PROJECT_ID
```
将 PROJECT_ID 替换为您要创建的 Google Cloud 项目的名称。
选择您创建的 Google Cloud 项目：
```
gcloud config set project PROJECT_ID
```
将 PROJECT_ID 替换为您的 Google Cloud 项目名称。

验证是否已为您的 Google Cloud 项目启用结算功能。

启用 Cloud Tasks、Compute Engine 和 Workflows API：

启用 API 所需的角色

如需启用 API，您需要拥有 Service Usage Admin IAM 角色 (roles/serviceusage.serviceUsageAdmin)，该角色包含 serviceusage.services.enable 权限。了解如何授予角色。

gcloud services enable cloudtasks.googleapis.com compute.googleapis.com workflows.googleapis.com

记下 Compute Engine 默认服务账号，因为您将在本教程中将其与工作流关联以进行测试。启用或使用包含 Compute Engine 的 Google Cloud 服务后，系统会自动创建此服务账号，其电子邮件地址格式如下：
```
PROJECT_NUMBER-compute@developer.gserviceaccount.com
```
将 PROJECT_NUMBER 替换为您的Google Cloud 项目编号。您可以通过运行以下命令找到项目编号：
```
gcloud projects describe PROJECT_ID --format='value(projectNumber)'
```
对于生产环境，我们强烈建议创建新的服务账号，并为其授予一个或多个 IAM 角色，这些角色包含所需的最小权限并遵循最小权限原则。

注意：
根据您的组织政策配置，默认服务账号可能会自动获得项目的 Editor 角色。我们强烈建议您通过强制执行 iam.automaticIamGrantsForDefaultServiceAccounts 组织政策限制条件来停用自动角色授予功能。如果您的组织是在 2024 年 5 月 3 日之后创建的，则默认情况下会强制执行此限制条件。

如果您停用自动角色授予功能，则必须决定向默认服务账号授予哪些角色，然后自行授予这些角色。

如果默认服务账号已具有 Editor 角色，我们建议您将 Editor 角色替换为权限较少的角色。如需安全地修改服务账号的角色，请使用 Policy Simulator 查看更改的影响，然后授予和撤消相应的角色。
将项目的 Workflows Invoker 角色 (roles/workflows.invoker) 授予 Compute Engine 默认服务账号，以便该账号有权触发您的工作流执行。
```
gcloud projects add-iam-policy-binding PROJECT_ID \
    --member=serviceAccount:PROJECT_NUMBER-compute@developer.gserviceaccount.com \
    --role=roles/workflows.invoker
```
替换以下内容：
- PROJECT_ID： Google Cloud 项目 ID
- PROJECT_NUMBER： Google Cloud 项目编号

将项目的 Cloud Tasks Enqueuer 角色 (roles/cloudtasks.enqueuer) 授予 Compute Engine 默认服务账号，以便该账号有权创建任务。

gcloud projects add-iam-policy-binding PROJECT_ID \
    --member=serviceAccount:PROJECT_NUMBER-compute@developer.gserviceaccount.com \
    --role=roles/cloudtasks.enqueuer

创建 Cloud Tasks 队列

创建可在父工作流中使用的 Cloud Tasks 队列，并允许您控制工作流执行速率。

控制台

在 Google Cloud 控制台中，前往 Cloud Tasks 页面：

转至 Cloud Tasks
点击 创建推送队列。
输入队列名称，即 queue-workflow-child。
在区域列表中，选择 us-central1（爱荷华）。
点击创建。

gcloud

QUEUE=queue-workflow-child
LOCATION=us-central1
gcloud tasks queues create $QUEUE --location=$LOCATION

创建和部署子工作流

子工作流可以接收和处理来自父工作流的数据。创建并部署一个执行以下操作的子工作流：

接收 iteration 作为实参
休眠 10 秒以模拟某些处理
成功执行后返回字符串

控制台

在 Google Cloud 控制台中，前往 Workflows 页面。

进入 Workflows
点击创建。
输入新工作流的名称 workflow-child。
在区域列表中，选择 us-central1（爱荷华）。
在服务账号列表中，选择 Compute Engine 默认服务账号。
点击下一步。

在工作流编辑器中，输入工作流的定义：

main:
  params: [args]
  steps:
    - init:
        assign:
          - iteration : ${args.iteration}
    - wait:
        call: sys.sleep
        args:
            seconds: 10
    - return_message:
        return: ${"Hello world"+iteration}

点击部署。

gcloud

为工作流创建源代码文件：
```
touch workflow-child.yaml
```

在文本编辑器中打开源代码文件，然后将以下工作流复制到该文件中。

main:
  params: [args]
  steps:
    - init:
        assign:
          - iteration : ${args.iteration}
    - wait:
        call: sys.sleep
        args:
            seconds: 10
    - return_message:
        return: ${"Hello world"+iteration}

部署工作流：

gcloud workflows deploy workflow-child \
    --source=workflow-child.yaml \
    --location=us-central1 \
    --service-account=PROJECT_NUMBER-compute@developer.gserviceaccount.com

创建和部署父工作流

父工作流使用 for 循环执行子工作流的多个分支。

复制定义父工作流的源代码：

main:
  steps:
    - init:
        assign:
          - project_id: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
          - project_number: ${sys.get_env("GOOGLE_CLOUD_PROJECT_NUMBER")}
          - location: ${sys.get_env("GOOGLE_CLOUD_LOCATION")}
          - workflow_child_name: "workflow-child"
          - queue_name: "queue-workflow-child"
    - enqueue_tasks_to_execute_child_workflow:
        for:
          value: iteration
          range: [1, 100]
          steps:
              - iterate:
                  assign:
                    - data:
                        iteration: ${iteration}
                    - exec:
                        # Encode object to JSON string in expression for workflow argument
                        argument: ${json.encode_to_string(data)}
              - create_task_to_execute_child_workflow:
                  call: googleapis.cloudtasks.v2.projects.locations.queues.tasks.create
                  args:
                      parent: ${"projects/" + project_id + "/locations/" + location + "/queues/" + queue_name}
                      body:
                        task:
                          httpRequest:
                            body: ${base64.encode(json.encode(exec))}
                            url: ${"https://workflowexecutions.googleapis.com/v1/projects/" + project_id + "/locations/" + location + "/workflows/" + workflow_child_name + "/executions"}
                            oauthToken:
                              serviceAccountEmail: ${project_number + "-compute@developer.gserviceaccount.com"}

工作流包括以下部分：

用于分配引用子工作流和 Cloud Tasks 队列名称的常量的映射。如需了解详情，请参阅地图。
一种 for 循环，用于以迭代方式调用子工作流。如需了解详情，请参阅迭代。
一种工作流步骤，用于创建大量任务并将其添加到 Cloud Tasks 队列中，以执行子工作流。如需了解详情，请参阅 Cloud Tasks API 连接器。

部署工作流：
控制台
1. 在 Google Cloud 控制台中，前往 Workflows 页面：
  
  进入 Workflows
2. 点击创建。
3. 输入新工作流的名称 workflow-parent。
4. 在区域列表中，选择 us-central1（爱荷华）。
5. 在服务账号列表中，选择 Compute Engine 默认服务账号。
6. 点击下一步。
7. 在工作流编辑器中，粘贴父工作流的定义。
8. 点击部署。
gcloud
1. 为工作流创建源代码文件：
  touch workflow-parent.yaml
2. 在文本编辑器中打开源代码文件，然后粘贴父工作流的定义。
3. 部署工作流：
  gcloud workflows deploy workflow-parent \ --source=workflow-parent.yaml \ --location=us-central1 \ --service-account=PROJECT_NUMBER-compute@developer.gserviceaccount.com

执行没有速率限制的父工作流

执行父工作流，以通过 Cloud Tasks 队列调用子工作流。执行大约需要 10 秒才能完成。

控制台

在 Google Cloud 控制台中，前往 Workflows 页面：

进入 Workflows
在工作流页面上，点击 workflow-parent 工作流以转到其详情页面。
在工作流详情页面上，点击执行。
再次点击执行。
在父工作流运行时，返回到Workflows页面，然后点击 workflow-child 工作流以转到其详情页面。
点击执行标签页。

您应该会看到子工作流的执行情况，这些执行大约在同一时间运行，类似于以下内容：

gcloud

执行工作流：

gcloud workflows run workflow-parent \
     --location=us-central1

如需验证工作流执行是否已触发，请列出最后四项执行：

gcloud workflows executions list workflow-child --limit=4

由于执行次数 (100) 低于 Workflows 并发限制，因此结果应与以下内容类似。如果您同时提交数千次执行，可能会出现配额问题。

NAME: projects/620278351741/locations/us-central1/workflows/workflow-child/executions/1570d06e-d133-4536-a859-b7b6a1a85524
STATE: ACTIVE
START_TIME: 2023-07-27T00:56:15.093934448Z
END_TIME:
NAME: projects/620278351741/locations/us-central1/workflows/workflow-child/executions/82724960-7d92-4961-aa2c-a0f0be46212c
STATE: ACTIVE
START_TIME: 2023-07-27T00:56:14.903007626Z
END_TIME:
NAME: projects/620278351741/locations/us-central1/workflows/workflow-child/executions/598126fb-37f9-45bc-91d8-aea7d795d702
STATE: ACTIVE
START_TIME: 2023-07-27T00:56:14.698260524Z
END_TIME:
NAME: projects/620278351741/locations/us-central1/workflows/workflow-child/executions/d2e9960b-f93f-4df4-a594-3e7e5c2be53f
STATE: ACTIVE
START_TIME: 2023-07-27T00:56:14.503818840Z
END_TIME:

您已创建并部署一个工作流，该工作流会调用子工作流 100 次。

执行具有速率限制的父工作流

对 Cloud Tasks 队列应用每秒一次的调度速率限制，然后执行父工作流。

控制台

在 Google Cloud 控制台中，前往 Cloud Tasks 页面：

转至 Cloud Tasks
点击您创建的 Cloud Tasks 队列 queue-workflow-child，然后点击修改队列。
在任务分派速率限制部分的最大分派数字段中，输入 1。
点击保存。
前往 Workflows 页面：

进入 Workflows
点击 workflow-parent 工作流以转到其详情页面。
在工作流详情页面上，点击执行。
再次点击执行。
在父工作流运行时，返回到Workflows页面，然后点击 workflow-child 工作流以转到其详情页面。
点击执行标签页。

您应该会看到子工作流的执行情况，以每秒一次请求的频率运行，类似于以下内容：

gcloud

更新 Cloud Tasks 队列，以应用每秒一次调度的速率限制：

gcloud tasks queues update $QUEUE \
    --max-dispatches-per-second=1 \
    --location=us-central1

执行工作流：

gcloud workflows run workflow-parent \
   --location=us-central1

如需验证工作流执行是否已触发，请列出最后四项执行：

gcloud workflows executions list workflow-child --limit=4

结果应类似于以下内容，每秒执行一次工作流：

NAME: projects/620278351741/locations/us-central1/workflows/workflow-child/executions/becf4957-9fb2-40d9-835d-0ff2dd0c1249
STATE: ACTIVE
START_TIME: 2023-07-27T01:07:24.446361457Z
END_TIME:
NAME: projects/620278351741/locations/us-central1/workflows/workflow-child/executions/6c1e7c4b-7ac6-4121-b351-1e2d56d10903
STATE: ACTIVE
START_TIME: 2023-07-27T01:07:23.448213989Z
END_TIME:
NAME: projects/620278351741/locations/us-central1/workflows/workflow-child/executions/f2ba5027-af40-4cd3-8cd0-b8033bcc6211
STATE: ACTIVE
START_TIME: 2023-07-27T01:07:22.431485914Z
END_TIME:
NAME: projects/620278351741/locations/us-central1/workflows/workflow-child/executions/ecc61ee5-fe87-49eb-8803-89dba929f6c8
STATE: ACTIVE
START_TIME: 2023-07-27T01:07:21.443466369Z
END_TIME:

您已成功部署一个工作流，该工作流以每秒一次的调度速率调用子工作流 100 次迭代。

清理

如果您为本教程创建了一个新项目，请删除项目。如果您使用的是现有项目，希望保留此项目且不保留本教程中添加的任何更改，请删除为教程创建的资源。

删除项目

为了避免产生费用，最简单的方法是删除您为本教程创建的项目。

要删除项目，请执行以下操作：

在 Google Cloud 控制台中，前往管理资源页面。
转到“管理资源”
在项目列表中，选择要删除的项目，然后点击删除。
在对话框中输入项目 ID，然后点击关闭以删除项目。

删除教程资源

删除在本教程中创建的工作流和 Cloud Tasks 资源：

控制台

如需删除工作流，请按以下步骤操作：
1. 在 Google Cloud 控制台中，前往 Workflows 页面：
  
  进入 Workflows
2. 从工作流列表中，点击工作流以转至其工作流详情页面。
3. 点击删除。
4. 输入工作流的名称，然后点击确认。
如需删除 Cloud Tasks 队列，请按以下步骤操作：
1. 在 Google Cloud 控制台中，前往 Cloud Tasks 页面：
  
  转至 Cloud Tasks
2. 选择要删除的队列的名称，然后点击删除队列。
3. 确认该操作。

gcloud

如需删除工作流，请运行以下命令：

gcloud workflows delete workflow-child
gcloud workflows delete workflow-parent

如需删除 Cloud Tasks 队列，请运行以下命令：
```
gcloud tasks queues delete queue-workflow-child
```

后续步骤

如需详细了解如何使用 Cloud Tasks 将工作流排入队列并异步执行，请参阅使用 Cloud Tasks 将工作流执行排入队列。
如需详细了解 Workflows 语法，请参阅 Workflows 语法参考文档。

使用 Cloud Tasks 队列缓冲工作流执行 使用集合让一切井井有条 根据您的偏好保存内容并对其进行分类。

目标

费用

准备工作

控制台

gcloud

创建 Cloud Tasks 队列

控制台

gcloud

创建和部署子工作流

控制台

gcloud

创建和部署父工作流

控制台

gcloud

执行没有速率限制的父工作流

控制台

gcloud

执行具有速率限制的父工作流

控制台

gcloud

清理

删除项目

删除教程资源

控制台

gcloud

后续步骤

使用 Cloud Tasks 队列缓冲工作流执行