配置 Cloud Tasks 队列

您可以在创建 Cloud Tasks 队列时或之后的任何时间配置该队列。该配置将应用于相应队列中的所有任务。

配置队列需要注意三个基本方面：

配置队列级路由

在队列级别配置的路由会覆盖在任务级别设置的路由。如果您想将 Cloud Tasks 用作目标服务前面的缓冲区，或者需要更改队列中所有任务的路由，则此方法非常有用。

队列级路由适用于：

  • 队列中的任务
  • 在设置队列级路由后添加到队列的任务

限制

队列级路由与 Cloud Key Management Service (Cloud KMS) 客户管理的加密密钥 (CMEK) 不兼容。如果启用了 CMEK，您将无法执行以下操作：

  • 在具有队列级路由的队列中创建任务
  • 应用队列级转送

为 HTTP 任务配置队列级路由

您可以在创建队列时或更新队列时配置队列以替换任务级路由。如需配置队列级路由，请将队列的 uriOverride 参数设置为首选路由。

如果您要将队列级路由作为对现有队列的更新来应用，请在应用更改之前暂停队列，并在应用更改后等待一分钟，然后再恢复队列。

  1. 通过运行以下命令暂停队列：

      gcloud tasks queues pause QUEUE_ID

    将 QUEUE_ID 替换为您的队列的 ID。

  2. 更新或移除队列级路由。

    • 如需更新队列级路由，请将 uriOverride 参数设置为更新后的路由。

    • 如需使用 REST 或 RPC API 移除队列级路由，请执行以下操作：

      • REST API：发送 patch 请求，该请求的载荷为空，且 updateMask 参数设置为 httpTarget。

      • RPC API：针对队列发送一个空载荷的 updateQueueRequest，并将 update_mask 参数设置为 http_target。

    以下示例使用 REST API 更新任务的路由目标主机：

    curl -X PATCH -d @- -i \
  -H "Authorization: Bearer ACCESS_TOKEN" \
  -H "Content-Type: application/json" \
  "https://cloudtasks.googleapis.com/v2/projects/PROJECT_ID/locations/LOCATION/queues/QUEUE_ID?updateMask=httpTarget.uriOverride" << EOF
{
  "httpTarget": {"uriOverride":{"host":"NEW_HOST"}}
}
EOF

    替换以下内容：

    • ACCESS_TOKEN：您的访问令牌。您可以在终端中运行以下命令来获取此信息：

      gcloud auth application-default login
gcloud auth application-default print-access-token

    • PROJECT_ID：您的 Google Cloud 项目的 ID。 您可以在终端中运行以下命令来获取此信息： 

      gcloud config get-value project

    • LOCATION：队列的位置。

    • NEW_HOST：您希望队列路由到的新主机。

  3. 等待 1 分钟。

    新配置最多可能需要一分钟才能生效。等待恢复队列有助于防止任务以旧配置进行调度。

  4. 运行以下命令以恢复队列：

    gcloud tasks queues resume QUEUE_ID

为 App Engine 任务配置队列级路由

如需为 App Engine 任务配置队列级路由，请将队列的 appEngineRoutingOverride 参数设置为您偏好的 App Engine 服务和版本。

  1. 设置队列级路由并覆盖任何任务级路由：

    gcloud tasks queues update QUEUE_ID \
    --routing-override=service:SERVICE,version:VERSION

    替换以下内容：

    • QUEUE_ID：队列 ID（其简称）。
    • SERVICE：负责任务处理的 App Engine 工作器服务。
    • VERSION：应用版本。

    例如，如果您设置了工作器服务来处理队列中的所有任务，则可以路由到该服务和默认版本：

    gcloud tasks queues update QUEUE_ID \
    --routing-override=service:SERVICE

  2. 运行以下命令，验证队列是否已配置成功：

    gcloud tasks queues describe QUEUE_ID --location=LOCATION

    将 LOCATION 替换为队列的位置。

    输出应类似如下所示：

    appEngineRoutingOverride:
  host: SERVICE.PROJECT_ID.appspot.com
  service: SERVICE
name: projects/PROJECT_ID/locations/LOCATION_ID/queues/QUEUE_ID
rateLimits:
  maxBurstSize: 100
  maxConcurrentDispatches: 1000
  maxDispatchesPerSecond: 500.0
retryConfig:
  maxAttempts: 100
  maxBackoff: 3600s
  maxDoublings: 16
  minBackoff: 0.100s
state: RUNNING

  3. 如需移除队列级路由，请运行以下命令：

    gcloud tasks queues update QUEUE_ID \
    --clear-routing-override

    移除队列级路由后，系统会将任务级路由应用于队列中的任务以及将来添加到队列中的任务。

定义速率限制

速率限制决定了队列可以分派任务的最大速率，无论分派是首次尝试执行任务还是重试。

  1. 运行以下命令，设置队列可调度的并发任务的最大速率和数量：

    gcloud tasks queues update QUEUE_ID \
    --max-dispatches-per-second=DISPATCH_RATE \
    --max-concurrent-dispatches=MAX_CONCURRENT_DISPATCHES

    替换以下内容：

    • QUEUE_ID：队列 ID（其简称）。
    • DISPATCH_RATE：派送率。这是存储桶中令牌的刷新速率。在任务流相对稳定的情况下，这相当于任务被分派的速率。
    • MAX_CONCURRENT_DISPATCHES：队列中可同时运行的任务数上限。

    例如，如果您在未设置任何参数的情况下创建了队列，则可以通过运行以下命令来更新最大并发任务数量：

    gcloud tasks queues update QUEUE_ID \
    --max-concurrent-dispatches=MAX_CONCURRENT_DISPATCHES

  2. 运行以下命令，验证队列是否已配置成功：

    gcloud tasks queues describe QUEUE_ID --location=LOCATION

    将 LOCATION 替换为队列的位置。

    输出应类似如下所示：

    name: projects/PROJECT_ID/locations/LOCATION_ID/queues/QUEUE_ID
rateLimits:
  maxBurstSize: 100
  maxConcurrentDispatches: MAX_CONCURRENT_DISPATCHES
  maxDispatchesPerSecond: 500.0
retryConfig:
  maxAttempts: 100
  maxBackoff: 3600s
  maxDoublings: 16
  minBackoff: 0.100s
state: RUNNING

定义队列处理速率的方法

您可以使用 Cloud Tasks API 或通过上传 queue.yaml 文件来定义队列处理速率。这两种方法都会导致队列使用相同的底层机制。

在这两种情况下，队列都使用令牌桶算法来控制任务执行速率。每个命名的队列都有一个用于存储令牌的存储桶。

应用每执行一个任务，系统就会从令牌桶中移除一个令牌。队列会继续处理任务，直到该队列的令牌桶中再也没有令牌为止。系统会按照您为队列指定的 max_dispatches_per_second 速率不断向令牌桶中补充填充新令牌。如果队列中有要处理的任务，并且队列的令牌桶中存在令牌，则系统同时处理的任务数量与令牌数量相同，最多不超过您设置的 max_concurrent_dispatches 值。

负载不平衡可能会导致存储桶中的令牌数量显着增加，这可能会在随后出现大量请求时导致处理量暴增。在这种情况下，您的队列可能会遇到超过 max_dispatches_per_second 速率的实际调度速率，这不仅消耗系统资源还会与用户服务请求争用资源。如果您使用队列管理基于相对较慢的下游服务 SLA 的调度速率，则会导致 HTTP 429（请求过多）或 HTTP 503（服务不可用）等错误。

  • 当您使用任何 Cloud Tasks API 方法时，您有两个字段来定义队列调度速率：

    • max_dispatches_per_second
    • max_concurrent_dispatches

    第三个字段 max_burst_size 由系统根据您为 max_dispatches_per_second 设置的值进行计算。如需了解详情，请参阅 RateLimits 消息。

  • 使用 queue.yaml 方法时，您可以设置以下全部三个元素：

    • max_concurrent_requests，等于 max_concurrent_dispatches
    • rate，等于 max_dispatches_per_second
    • bucket_size，等于 max_burst_size

在大多数情况下，使用 Cloud Tasks API 方法并让系统设置 max_burst_size 可以非常高效地管理请求爆发期。但在某些情况下，具体来说是当所需速率较慢时，可以使用 queue.yaml 方法将 bucket_size 手动设置为较小的值，或者通过 Cloud Tasks 将 max_concurrent_dispatches 设置为较小的值 API 来为您提供更多控制权。

设置重试参数

如果任务未成功完成，Cloud Tasks 将根据您设置的参数，使用指数退避算法重试该任务。

  1. 运行以下命令，指定队列中失败任务的最大重试次数，设置重试尝试的时间限制，以及控制重试之间的时间间隔：

    gcloud tasks queues update QUEUE_ID \
    --max-attempts=MAX_ATTEMPTS \
    --max-retry-duration=MAX_RETRY_DURATION \
    --min-backoff=MIN_INTERVAL \
    --max-backoff=MAX_INTERVAL \
    --max-doublings=MAX_DOUBLINGS

    替换以下内容：

    • QUEUE_ID：队列 ID（其简称）。
    • MAX_ATTEMPTS：任务可以尝试的最大次数，包括第一次尝试。您可以将此标志设置为 -1 来允许进行无限次重试。请注意，即使 MAX_ATTEMPTS 设置为 -1，MAX_RETRY_DURATION 仍会应用。
    • MAX_RETRY_DURATION：重试失败任务的最长时间（从首次尝试任务时算起）。该值必须是以“s”结尾的字符串，如 5s。如果设置为 0，则任务年龄不受限制。请注意，即使 MAX_RETRY_DURATION 设置为 0，MAX_ATTEMPTS 仍会应用。
    • MIN_INTERVAL：重试尝试之间的最短等待时间。该值必须是以“s”结尾的字符串，如 5s。
    • MAX_INTERVAL：重试尝试之间的最长等待时间。该值必须是以“s”结尾的字符串，如 5s。

    • MAX_DOUBLINGS：在增加量变为常量之前，失败任务重试之间的时间间隔将加倍的最大次数。任务的重试间隔从 MIN_INTERVAL 开始，然后加倍 MAX_DOUBLINGS 次，接着呈线性增加，最后以 MAX_INTERVAL 为间隔重试，直到达到 MAX_ATTEMPTS 次。

      例如，如果 MIN_INTERVAL 为 10s、MAX_INTERVAL 为 300s 且 MAX_DOUBLINGS 为 3，则重试间隔将加倍 3 次，线性增加 2^3 * 10 秒，然后以 MAX_INTERVAL 的间隔重试，直到任务尝试次数达到 MAX_ATTEMPTS 次：10 秒、20 秒、40 秒、80 秒、160 秒、240 秒、300 秒、300 秒，依此类推。

    如需了解更多参数详情，请参阅Queue 资源的 RetryConfig 设置。

  2. 运行以下命令，验证队列是否已配置成功：

    gcloud tasks queues describe QUEUE_ID --location=LOCATION

    将 LOCATION 替换为队列的位置。

    输出应包含您设置的重试参数。

后续步骤