将推送队列迁移到 Cloud Tasks (Python)

本页介绍如何将推送队列代码从任务队列迁移到 Cloud Tasks。Cloud Tasks 现在是使用 App Engine 推送队列的首选方式。

使用 Cloud Tasks 时访问的服务与使用任务队列 RPC API 时相同。这意味着您无需重新创建现有的推送队列和推送任务。但是，您必须迁移用于创建推送队列或推送任务或者用于与这两者交互的代码，才能使用 Cloud Tasks API。

您可以使用 Cloud Tasks REST 和 RPC API、Cloud Tasks 客户端库、Google Cloud CLI 和 Google Cloud 控制台创建推送队列和推送任务并与之交互。本页提供了使用 gcloud CLI 和 Cloud Tasks 客户端库的示例。

在 Cloud Tasks 中，所有队列均作为推送队列运行。在本指南的其余部分和 Cloud Tasks 文档中，“队列”和“推送队列”这两个术语含义相同。类似地，“任务”和“推送任务”这两个术语含义也相同。

Cloud Tasks 不提供的功能

Cloud Tasks 不提供以下功能：

对 Datastore 事务中的任务进行排队
使用延迟任务库而非工作器服务
在多租户应用中处理任务
使用本地开发服务器进行模拟
异步添加任务

价格和配额

将推送队列迁移到 Cloud Tasks 可能会影响应用的价格和配额。

价格

Cloud Tasks 有自己的价格。与任务队列一样，向 App Engine 应用发送带有任务的请求会导致应用产生费用。

配额

Cloud Tasks 配额不同于任务队列的配额。与任务队列一样，从 Cloud Tasks 向 App Engine 应用发送请求可能会影响 App Engine 请求配额。

迁移之前

如果尚未设置，请设置 Python 开发环境以使用与 Google Cloud 兼容的 Python 版本，并安装用于创建独立 Python 环境的测试工具。

以下部分讨论了在将推送队列迁移到 Cloud Tasks 之前的设置步骤。

迁移拉取队列

在开始之前，请先迁移拉取队列，然后再按照本指南中的说明迁移推送队列。不建议在迁移推送队列后迁移拉取队列，因为必须使用 queue.yaml 文件这一要求可能会导致 Cloud Tasks 出现意外行为。

保护队列配置

一旦开始迁移到 Cloud Tasks，就不建议修改 queue.yaml 文件，因为这样做会导致意外行为。按照以下步骤保护您的队列配置，以免被 queue.yaml 文件修改。

配置 gcloud CLI，使之在将来的部署中省略 queue.yaml 文件。

将您的 queue.yaml 文件添加到 .gcloudignore 文件。如需检查是否已有 .gcloudignore 文件，您可以在终端中从应用的顶级目录运行以下命令。如果该文件存在，此命令将输出文件名。
```
ls -a | grep .gcloudignore
```
如需详细了解 .gcloudignore 文件，请参阅 .gcloudignore 参考文档。
限制对您的 queue.yaml 文件的权限。

遵循保护队列配置指南中所述的最佳做法。
了解 Cloud Tasks 和 queue.yaml 文件（可选）。

使用 Cloud Tasks API 管理队列配置时，部署 queue.yaml 文件会覆盖由 Cloud Tasks 设置的配置，这会导致意外行为。如需了解详情，请阅读使用队列管理与使用 queue.yaml 的对比。

启用 Cloud Tasks API

如需启用 Cloud Tasks API，请点击 API 库中 Cloud Tasks API 上的启用。如果您看到的是管理按钮，而不是启用按钮，则说明您之前已为项目启用了 Cloud Tasks API，因而无需再次执行此操作。

对应用进行身份验证以访问 Cloud Tasks API

您必须对应用进行身份验证，然后才能访问 Cloud Tasks API。本部分介绍了针对两种不同使用场景的身份验证。

如需在本地开发或测试您的应用，建议您使用服务账号。如需了解如何设置服务账号并将其与应用关联，请参阅手动获取和提供服务账号凭据。

如需在 App Engine 上部署应用，您不需要提供任何新的身份验证。应用默认凭据 (ADC) 会推断 App Engine 应用的身份验证详细信息。

下载 gcloud CLI

下载并安装 gcloud CLI 以便将 gcloud CLI 与 Cloud Tasks API 搭配使用（如果您之前未安装过）。如果您已经安装 gcloud CLI，请从终端运行以下命令。

gcloud components update

导入 Cloud 客户端库

如需将 Cloud Tasks 客户端库与 App Engine 应用搭配使用，请执行以下操作：

更新 app.yaml 文件。按照适用于您的 Python 版本的说明操作：
Python 2
对于 Python 2 应用，请添加最新版本的 grpcio 和 setuptools 库。

下面是一个 app.yaml 文件示例：
```
runtime: python27
threadsafe: yes
api_version: 1

libraries:
- name: grpcio
  version: latest
- name: setuptools
  version: latest
```
Python 3
对于 Python 3 应用，请在 app.yaml 文件中使用受支持的 Python 3 版本指定 runtime 元素。例如：
```
runtime: python310 # or another support version
```
Python 3 运行时会自动安装库，因此您无需指定过往 Python 2 运行时中的内置库。如果您的 Python 3 应用在迁移时使用其他旧版捆绑服务，您可以继续指定必要的内置库。否则，您可以删除 app.yaml 文件中的不必要行。
更新 requirements.txt 文件。按照适用于您的 Python 版本的说明操作：
Python 2
将适用于 Cloud Tasks 的 Cloud 客户端库添加到 requirements.txt 文件中的依赖项列表。
```
google-cloud-tasks
```
然后，运行 pip install -t lib -r requirements.txt 以更新应用的可用库列表。
Python 3
将适用于 Cloud Tasks 的 Cloud 客户端库添加到 requirements.txt 文件中的依赖项列表。
```
google-cloud-tasks
```
App Engine 会在 Python 3 运行时的应用部署期间自动安装这些依赖项，因此如果存在 lib 文件夹，请删除它。
对于 Python 2 应用，如果您的应用使用内置库或复制的库，您必须在 appengine_config.py 文件（与 app.yaml 文件位于同一文件夹）中指定这些路径：
```
import pkg_resources
from google.appengine.ext import vendor

# Set PATH to your libraries folder.
PATH = 'lib'
# Add libraries installed in the PATH folder.
vendor.add(PATH)
# Add libraries to pkg_resources working set to find the distribution.
pkg_resources.working_set.add_entry(PATH)
```
appengine_config.py 文件假设当前工作目录是 lib 文件夹所在的位置。在某些情况下，比如单元测试，当前工作目录可能有所不同。为避免出错，您可以使用以下命令，将完整路径明确传递到 lib 文件夹：
```
import os
path = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'lib')
```
在使用 Task Queues API 中的推送队列的任何文件中导入 Cloud Tasks 客户端库：
```
from google.cloud import tasks
```
在您成功将用于创建推送队列或与之交互的所有代码完整迁移到 Cloud Tasks 后，请移除导入 Task Queues API 的语句，例如 from google.appengine.api import taskqueue。

创建和管理队列

本部分介绍如何使用 Cloud Tasks API 创建和管理队列。

使用 Cloud Tasks 时，您不需要使用 queue.yaml 文件来创建或管理队列，而应使用 Cloud Tasks API。不建议同时使用 queue.yaml 文件和 Cloud Tasks API，但这可能是从任务队列迁移到 Cloud Tasks 的必要步骤，具体取决于您的应用。请参阅使用队列管理与使用 queue.yaml 的对比，了解最佳做法。

创建队列

如果您的应用以编程方式创建队列，或者您要通过命令行创建其他队列，请阅读本部分。

在 Cloud Tasks 中，队列名称的格式为 projects/PROJECT_ID/locations/LOCATION_ID/queues/QUEUE_ID。队列名称的 LOCATION_ID 部分对应于 Google Cloud 地区。队列名称的 QUEUE_ID 部分等同于任务队列的队列 name 字段。队列名称是在队列创建过程中根据您指定的项目、地区和 QUEUE_ID 生成的。

通常，队列位置（即区域）必须与应用的区域相同。此规则的两个例外情况适用于使用 europe-west 区域的应用和使用 us-central 区域的应用。在 Cloud Tasks 中，这些地区分别称为 europe-west1 和 us-central1。

您可以在创建队列的过程中指定可选的队列配置，但也可以在创建队列后对其进行更新。

您无需重新创建现有队列。请阅读本指南的相关部分，迁移与现有队列交互的代码。

重复使用队列名称

删除队列后，您必须等待 7 天才能在同一项目和位置（即区域）中创建具有相同队列 ID 的队列。

以下示例会使用 Cloud Tasks 创建两个队列。第一个队列的队列 ID 为 queue-blue，并且配置为以 5/s 的速率将所有任务发送到服务 task-module 的版本 v2。第二个队列的队列 ID 为 queue-red，并以 1/s 的速率分配任务。这两者都是在 us-central1 位置中的项目（项目 ID 为 my-project-id）上创建。这是与任务队列中的创建队列等效的 Cloud Tasks 功能。

gcloud

gcloud CLI 会根据 gcloud CLI 配置推断项目和位置。

gcloud tasks queues create queue-blue \
--max-dispatches-per-second=5 \
--routing-override=service:task-module,version:v2

gcloud tasks queues create queue-red \
--max-dispatches-per-second=1

客户端库

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue_blue_name = 'queue-blue'
# queue_red_name = 'queue-red'

parent = f"projects/{project}/locations/{location}"

queue_blue = {
    "name": client.queue_path(project, location, queue_blue_name),
    "rate_limits": {"max_dispatches_per_second": 5},
    "app_engine_routing_override": {"version": "v2", "service": "task-module"},
}

queue_red = {
    "name": client.queue_path(project, location, queue_red_name),
    "rate_limits": {"max_dispatches_per_second": 1},
}

queues = [queue_blue, queue_red]
for queue in queues:
    response = client.create_queue(parent=parent, queue=queue)
    print(response)

如需了解详情，请参阅 Cloud Tasks 参考文档创建 Cloud Tasks 队列。

设置队列处理速率

下表列出了任务队列与 Cloud Tasks 中的不同字段。

任务队列字段	Cloud Tasks 字段	说明
`rate`	`max_dispatches_per_second`	从队列中分派任务的最大速率
`max_concurrent_requests`	`max_concurrent_dispatches`	可从队列中分派的并发任务数上限
`bucket_size`	`max_burst_size`	Cloud Tasks 会计算 get-only 属性 `max_burst_size`，用于根据 `max_dispatches_per_second` 的值限制队列中任务的处理速度。此字段允许队列具有较高的速率，以便任务在排入队列后很快便能予以处理，但如果短时间内有很多任务排入队列，则仍会限制资源使用率。对于使用 `queue.yaml` 文件创建或更新的 App Engine 队列，`max_burst_size` 最初等于 `bucket_size`。但是，如果稍后使用任何 Cloud Tasks 接口将队列传递到 `update` 命令，则无论 `max_dispatches_per_second` 是否已更新，系统都会根据 `max_dispatches_per_second` 的值重置 `max_burst_size`。
`total_storage_limit`	在 Cloud Tasks 中已弃用	Cloud Tasks 不支持设置自定义存储限制

您可以在创建队列时设置队列处理速率，也可以在事后更新它。以下示例会使用 Cloud Tasks 为已创建的名为 queue-blue 的队列设置处理速率。如果 queue-blue 是使用 queue.yaml 文件创建或配置的，则以下示例会根据 max_dispatches_per_second 值 20 重置 max_burst_size。这是与任务队列中的设置队列处理速率等效的 Cloud Tasks 功能。

gcloud

gcloud tasks queues update queue-blue \
--max-dispatches-per-second=20 \
--max-concurrent-dispatches=10

客户端库

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'queue-blue'

# Get queue object
queue_path = client.queue_path(project, location, queue)
queue = client.get_queue(name=queue_path)

# Update queue object
queue.rate_limits.max_dispatches_per_second = 20
queue.rate_limits.max_concurrent_dispatches = 10

response = client.update_queue(queue=queue)
print(response)

如需了解详情，请参阅定义速率限制。

停用和恢复队列

Cloud Tasks 使用“暂停”这一术语的方式与任务队列使用“停用”这一术语的方式相同。暂停某个队列会使该队列中的任务停止执行，直到该队列恢复为止。但是，您可以继续将任务添加到已暂停的队列中。Cloud Tasks 使用“恢复”这一术语的方式与任务队列使用该术语的方式相同。

以下示例会暂停队列 ID 为 queue1 的队列。这是与任务队列中的停用队列等效的 Cloud Tasks 功能。

gcloud

gcloud tasks queues pause 

queue1

客户端库

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'queue1'

queue_path = client.queue_path(project, location, queue)
response = client.pause_queue(name=queue_path)

如需了解详情，请参阅 Cloud Tasks 参考文档暂停队列。

删除队列

删除队列后，您必须等待 7 天才能创建同名的队列。如果您无法等待 7 天，请考虑清除队列中的所有任务，并重新配置队列。

以下示例会删除队列 ID 为 queue1 的队列。这是与任务队列中的删除队列等效的 Cloud Tasks 功能。

gcloud

gcloud tasks queues delete 

queue1

客户端库

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'queue1'

queue_path = client.queue_path(project, location, queue)
response = client.delete_queue(name=queue_path)

如需了解详情，请参阅 Cloud Tasks 参考文档删除队列。

创建和管理任务

本部分介绍如何使用 Cloud Tasks API 创建和管理任务。

创建任务

下表列出了任务队列与 Cloud Tasks 中的不同字段。

任务队列字段	Cloud Tasks 字段	说明
Cloud Tasks 中的新功能	`app_engine_http_request`	创建一个针对 App Engine 服务发出的请求。这些任务称为 App Engine 任务。
`method`	`http_method`	指定请求方法；例如 `POST`
`url`	`relative_uri`	指定任务处理程序。注意最后一个字母的区别：`i`（统一资源标识符），而不是 `l`（统一资源定位符）
`target`	`app_engine_routing`	可选。为 App Engine 任务指定 App Engine `service`、`version` 和 `instance`。如果未设置，则使用默认服务、版本和实例。

以下示例创建一个任务，该任务通过 /update_counter 处理程序路由到名为 worker 的 App Engine 服务。这是与任务队列中的创建任务等效的 Cloud Tasks 功能。

gcloud

gcloud tasks create-app-engine-task --queue=default \
--method=POST --relative-uri=/update_counter --routing=service:worker \
--body-content=10

客户端库

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'default'
amount = 10

parent = client.queue_path(project, location, queue)

task = {
    "app_engine_http_request": {
        "http_method": tasks.HttpMethod.POST,
        "relative_uri": "/update_counter",
        "app_engine_routing": {"service": "worker"},
        "body": str(amount).encode(),
    }
}

response = client.create_task(parent=parent, task=task)
eta = response.schedule_time.strftime("%m/%d/%Y, %H:%M:%S")
print(f"Task {response.name} enqueued, ETA {eta}.")

如需了解详情，请阅读 Cloud Tasks 参考文档创建 App Engine 任务。

指定目标服务和路由

为 App Engine 任务指定 App Engine 目标服务、版本和实例是可选操作。默认情况下，App Engine 任务会路由到尝试执行该任务时的默认设置服务、版本和实例。

在创建任务期间设置该任务的 app_engine_routing 属性，为您的任务指定不同的 App Engine 服务、版本或实例。

如需将指定队列上的所有任务路由到同一 App Engine 服务、版本和实例，您可以在队列上设置 app_engine_routing_override 属性。

如需了解详情，请参阅 Cloud Tasks 参考文档配置路由。

将数据传递给处理程序

与任务队列一样，您可以使用 Cloud Tasks 以两种方式将数据传递给处理程序。您可以在相对 URI 中将数据作为查询参数传递，也可以使用 HTTP 方法 POST 或 PUT 在请求正文中传递数据。

Cloud Tasks 使用“正文”这一术语的方式与任务队列使用“载荷”这一术语的方式相同。在 Cloud Tasks 中，默认正文内容类型是八位字节流而不是纯文本。您可以通过在标头中指定正文内容类型来设置它。

以下示例以两种不同的方式将键传递给处理程序 /update_counter。这是与任务队列中的将数据传递给处理程序等效的 Cloud Tasks 功能。

控制台

gcloud tasks create-app-engine-task --queue=default --method=GET  \
--relative-uri=

/update_counter

?key=blue --routing=service:worker

gcloud tasks create-app-engine-task --queue=default --method=POST \
--relative-uri=

/update_counter

 --routing=service:worker \
--body-content="{'key': 'blue'}"

客户端库

import json

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'default'

parent = client.queue_path(project, location, queue)

task1 = {
    "app_engine_http_request": {
        "http_method": tasks.HttpMethod.POST,
        "relative_uri": "/update_counter?key=blue",
        "app_engine_routing": {"service": "worker"},
    }
}

task2 = {
    "app_engine_http_request": {
        "http_method": tasks.HttpMethod.POST,
        "relative_uri": "/update_counter",
        "app_engine_routing": {"service": "worker"},
        "headers": {"Content-Type": "application/json"},
        "body": json.dumps({"key": "blue"}).encode(),
    }
}

response = client.create_task(parent=parent, task=task1)
print(response)
response = client.create_task(parent=parent, task=task2)
print(response)

为任务命名

您可以视需要指定任务名称。如果您未指定任务名称，Cloud Tasks 会通过生成任务 ID 并根据您在任务创建期间指定的队列推理项目和位置（即区域）来为您构造任务名称。

任务名称的格式为 projects/PROJECT_ID/locations/LOCATION_ID/queues/QUEUE_ID/tasks/TASK_ID。任务名称的 TASK_ID 部分相当于任务队列任务 name 字段。

重复使用任务名称

您必须等待一段时间才能重复使用任务名称。执行此操作之前必须等待的时间长短取决于是否在 Cloud Tasks 或任务队列中创建了分派任务的队列。

对于使用任务队列（包括默认队列）创建的队列上的任务，您必须在原始任务删除或执行后等待大约 9 天。对于使用 Cloud Tasks 创建的队列上的任务，您必须在原始任务删除或执行后等待大约 1 小时。

以下示例会创建一个将 TASK_ID 设置为 first-try 的任务，并将其添加到默认队列。这是与任务队列中的为任务命名等效的 Cloud Tasks 功能。

gcloud

gcloud CLI 通过根据配置推断项目和位置来构造任务名称。

gcloud tasks create-app-engine-task first-try --queue=default \
--method=GET --relative-uri=

/url/path

客户端库

使用客户端库时，如果要指定 TASK_ID，则必须指定完整任务名称。项目和位置必须与添加任务的目标队列的项目和位置匹配。

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'default'
# task_name = 'first-try'

parent = client.queue_path(project, location, queue)

task = {
    "name": client.task_path(project, location, queue, task_name),
    "app_engine_http_request": {
        "http_method": tasks.HttpMethod.GET,
        "relative_uri": "/url/path",
    },
}
response = client.create_task(parent=parent, task=task)
print(response)

重试之前执行失败的任务

您可以在创建队列过程中在队列上设置任务重试配置，也可以通过更新队列来设置。下表列出了任务队列字段和相应的 Cloud Tasks 字段。

任务队列字段	Cloud Tasks 字段
`task_retry_limit`	`max_attempts`
`task_age_limit`	`max_retry_duration`
`min_backoff_seconds`	`min_backoff`
`max_backoff_seconds`	`max_backoff`
`max_doublings`	`max_doublings`

任务特定的重试参数

在任务队列中配置的任务特定重试参数可在 Cloud Tasks 中使用，但您无法修改这些参数，也无法在新任务中设置这些参数。如需更改具有任务特定重试参数的任务的重试参数，请使用具有所需重试参数的 Cloud Tasks 队列重新创建该任务。

以下示例演示了各种重试方案：

在 fooqueue 中，从第一次尝试执行算起，任务最多重试七次，最长持续两天。两个限制都达到后，任务将永远失败。
在 barqueue 中，App Engine 会尝试重试任务，线性地增加每次重试之间的时间间隔，直到达到最大退避时间，然后以最大时间间隔无限次重试（因此，请求之间的时间间隔为 10 秒、20 秒、30 秒、…、190 秒、200 秒、200 秒、…）。
在 bazqueue 中，重试时间间隔从 10 秒开始，之后加倍三次，然后线性增加，最后以最大时间间隔无限期重试（因此，请求之间的时间间隔为 10 秒、20 秒、40 秒、80 秒、160 秒、240 秒、300 秒、300 秒、…）。

这是与任务队列中的重试任务等效的 Cloud Tasks 功能。

gcloud

设置指定秒数的选项时，必须在整数后加 s（例如 200s，而非 200）。

gcloud tasks queues create fooqueue \
--max-attempts=7 \
--max-retry-duration=172800s  #2*60*60*24 seconds in 2 days

gcloud tasks queues create barqueue \
--min-backoff=10s \
--max-backoff=200s \
--max-doublings=0

gcloud tasks queues create bazqueue \
--min-backoff=10s \
--max-backoff=300s \
--max-doublings=3

客户端库

from google.protobuf import duration_pb2

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# fooqueue = 'fooqueue'
# barqueue = 'barqueue'
# bazqueue = 'bazqueue'

parent = f"projects/{project}/locations/{location}"

max_retry = duration_pb2.Duration()
max_retry.seconds = 2 * 60 * 60 * 24

foo = {
    "name": client.queue_path(project, location, fooqueue),
    "rate_limits": {"max_dispatches_per_second": 1},
    "retry_config": {"max_attempts": 7, "max_retry_duration": max_retry},
}

min = duration_pb2.Duration()
min.seconds = 10

max = duration_pb2.Duration()
max.seconds = 200

bar = {
    "name": client.queue_path(project, location, barqueue),
    "rate_limits": {"max_dispatches_per_second": 1},
    "retry_config": {"min_backoff": min, "max_backoff": max, "max_doublings": 0},
}

max.seconds = 300
baz = {
    "name": client.queue_path(project, location, bazqueue),
    "rate_limits": {"max_dispatches_per_second": 1},
    "retry_config": {"min_backoff": min, "max_backoff": max, "max_doublings": 3},
}

queues = [foo, bar, baz]
for queue in queues:
    response = client.create_queue(parent=parent, queue=queue)
    print(response)

如需了解详情，请参阅 Cloud Tasks 参考文档设置重试参数。

从队列中删除任务

删除任务时，如果任务位于使用 queue.yaml 文件创建的队列中，您必须等待 9 天才能创建具有相同名称的任务；如果任务位于使用 Cloud Tasks 创建的队列中，则必须等待 1 小时。

以下示例会从队列 ID 为 queue1 的队列中删除任务 ID 为 foo 的任务。这是与任务队列中的删除任务等效的 Cloud Tasks 功能。

gcloud

系统会根据 gcloud CLI 默认项目推断任务项目和位置。

gcloud tasks delete foo --queue=queue1

客户端库

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'queue1'

task_path = client.task_path(project, location, queue, "foo")
response = client.delete_task(name=task_path)

如需了解详情，请参阅 Cloud Tasks 参考文档从队列中删除任务。

清除任务

以下示例会从队列 ID 为 queue1 的队列中清除所有任务。这是与任务队列中的清除任务等效的 Cloud Tasks 功能。

gcloud

系统会根据 gcloud CLI 默认项目推断队列项目和位置。

gcloud tasks queues purge 

queue1

客户端库

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'queue1'

queue_path = client.queue_path(project, location, queue)
response = client.purge_queue(name=queue_path)

如需了解详情，请参阅 Cloud Tasks 参考文档从队列中清除所有任务。

后续步骤

Cloud Tasks 文档
Cloud Tasks 客户端库
Cloud Tasks REST 参考文档概览
Cloud Tasks RPC 参考文档概览
如需查看教程，请参阅从 App Engine 任务队列推送队列迁移到 Cloud Tasks Codelab。

将推送队列迁移到 Cloud Tasks (Python)

Cloud Tasks 不提供的功能

价格和配额

价格

配额

迁移之前

迁移拉取队列

保护队列配置

启用 Cloud Tasks API

对应用进行身份验证以访问 Cloud Tasks API

下载 gcloud CLI

导入 Cloud 客户端库

Python 2

Python 3

Python 2

Python 3

创建和管理队列

创建队列

重复使用队列名称

gcloud

客户端库

设置队列处理速率

gcloud

客户端库

停用和恢复队列

gcloud

客户端库

删除队列

gcloud

客户端库

创建和管理任务

创建任务

gcloud

客户端库

指定目标服务和路由

将数据传递给处理程序

控制台

客户端库

为任务命名

重复使用任务名称

gcloud

客户端库

重试之前执行失败的任务

任务特定的重试参数

gcloud

客户端库

从队列中删除任务

gcloud

客户端库

清除任务

gcloud

客户端库

后续步骤