对作业请求进行去重

Cloud Scheduler 在 HTTP 和 App Engine 请求中包含作业标识和计划时间 HTTP 标头，可让您对重试的执行进行去重，并使目标处理程序幂等。由于 Cloud Scheduler 提供至少一次传送，因此在发生超时或暂时性网络中断时，对请求进行去重可防止处理程序或下游 Google Cloud API 多次处理同一预定运行。

请求去重的工作原理

在以下情况下，同一计划运行可能会出现重复的作业请求：

  • 长时间运行的目标：目标服务的响应时间超过作业的尝试截止时间 (attemptDeadline)，导致 Cloud Scheduler 在首次尝试仍在运行时超时并重试作业。对于 HTTP 目标，attemptDeadline 默认为 3 分钟，最长可配置为 30 分钟。使 attemptDeadline 与目标的超时时间保持一致，或者如果执行时间可能超过 30 分钟，则异步处理长时间运行的工作。
  • 网络确认丢失或临时代理重试：目标接收并处理请求，但临时网络中断或代理断开连接导致 Cloud Scheduler 无法接收 HTTP 2xx 响应。即使您将 retryCount 设置为 0，至少一次传送也意味着目标在瞬时传输故障期间仍可能收到重复请求。

如需使作业处理程序具有幂等性并对重试的请求进行去重，请使用 Cloud Scheduler 自动包含在对 HTTP 和 App Engine 目标的请求中的 HTTP 标头：

标题 说明
X-CloudScheduler-JobName projects/PROJECT_ID/locations/LOCATION_ID/jobs/JOB_ID 中的简短作业标识符 (JOB_ID)。
X-CloudScheduler-ScheduleTime

对于以 unix-cron 格式指定的作业，计划执行时间为相对于世界协调时间 (UTC) 的偏移量，格式符合 RFC 3339，例如 2026-10-16T12:00:00Z 或 2026-10-16T05:00:00-07:00。对于同一预定运行的所有重试尝试，此值保持不变。

对于使用替代 groc 时间表格式（例如 every 5 minutes）的作业，此标头不一定存在。如果您手动触发作业（例如，通过运行 gcloud scheduler jobs run 或在 Google Cloud 控制台中点击强制运行作业），则此标头会使用下一次计划运行的时间戳，或者可能会被省略。因此，基于标头的去重功能会跳过即将进行的预定运行，以及在该预定时间之前触发的所有其他手动运行。

在目标端点上要求进行身份验证

X-CloudScheduler-* 标头未经身份验证，可由任何调用方设置。请勿使用这些标头进行身份验证。而是根据目标类型保护端点：

使用不区分大小写的标头查找

标头大小写可能因 HTTP 版本和基础架构而异（例如，HTTP/1.1 中的 X-Cloudscheduler-Jobname 或 Cloud Run 等 HTTP/2 环境中的小写 x-cloudscheduler-jobname 和 x-cloudscheduler-scheduletime）。在处理程序中，请始终使用不区分大小写的标头查找。

消除跨项目或区域的作业 ID 歧义

Cloud Scheduler 不会自动发送项目 ID 或区域标头。单个处理程序端点可能会收到来自共享同一 JOB_ID 的多个项目或区域的请求。在这种情况下，请在 HttpTarget.headers、AppEngineHttpTarget.headers 或目标网址路径或查询参数中传递项目 ID 或区域，并将这些值包含在去重键中。

在 HTTP 处理程序中对请求进行去重

在处理程序中实现请求去重之前，请先完成以下设置前提条件：

  • 以 unix-cron 格式（例如 */5 * * * *，而不是 every 5 minutes）配置作业的时间表，以便 Cloud Scheduler 发送 X-CloudScheduler-ScheduleTime 标头。
  • 设置支持事务或原子条件写入的永久性数据存储区，例如 Firestore、Spanner 或 Memorystore。在集合或表（例如 expire_at 字段）上配置存留时间 (TTL) 政策，以便在作业的重试窗口 (maxRetryDuration) 结束后删除旧的去重记录。Firestore TTL 会异步删除过期记录，通常在 24-72 小时内完成。如需让重试在 TTL 清理之前从崩溃的实例中回收运行，还需存储租约到期时间戳（例如 lease_expires，设置为至少为目标的请求超时和 attemptDeadline），并在事务中对其进行评估。

如需在 HTTP 处理程序中仅处理一次每个预定运行，请完成以下步骤：

  1. 从传入请求标头中读取 X-CloudScheduler-JobName 和 X-CloudScheduler-ScheduleTime，并将其哈希处理为固定长度的重复信息删除键 (run_id)。
  2. 在数据存储区中运行事务，以检查并声明去重键的记录：
    • 如果记录存在且 status 为 completed，则立即返回 HTTP 200 OK（或 204 No Content）状态代码，而不运行作业逻辑，以便 Cloud Scheduler 停止重试。
    • 如果记录存在、status 为 in_progress 且租约 (lease_expires) 尚未过期，则返回可重试的错误状态代码（例如 409 Conflict），以便 Cloud Scheduler 在进行中的尝试失败时保持其重试政策处于有效状态。避免从 App Engine 目标返回 503 Service Unavailable，因为 App Engine 会将 503 视为系统错误，并重试请求，而不会将其计入 retryCount。
    • 如果该记录不存在或其租用已过期，请写入该记录，并将 status 设置为 in_progress、唯一尝试 ID (attempt_id)、租用到期时间戳 (lease_expires) 和 TTL 到期时间戳 (expire_at)。

  3. 声明记录后，运行作业逻辑。在事务中，先验证记录的 attempt_id 是否仍与当前尝试匹配，然后再修改记录：

    • 如果作业逻辑成功，请将 status 更新为 completed 并返回 HTTP 200 OK 状态代码。
    • 如果作业逻辑失败并返回可重试的错误，请先删除重复信息删除记录，然后再返回 5xx 状态代码，以便 Cloud Scheduler 无需等待租约到期即可重试运行。

HTTP 处理程序示例

以下 Python 示例（适用于 Cloud Run 或 Cloud Run functions）使用具有租约的 Firestore 事务来声明运行并对重试的作业请求进行去重。

此示例不适用于生产环境。在部署之前，请根据需要查看并调整：

import datetime
import hashlib
import uuid

import flask
import functions_framework
from google.cloud import firestore

# Longer than the work takes, and at least the job's attempt deadline.
LEASE = datetime.timedelta(minutes=5)
# Keep records past the job's retry window. A TTL policy on expire_at deletes them.
RETENTION = datetime.timedelta(days=1)

db = firestore.Client()


def run_job_logic() -> None:
    """Does the scheduled work. Replace the body with your own task."""


@firestore.transactional
def claim_run(
    transaction: firestore.Transaction,
    run_ref: firestore.DocumentReference,
    attempt_id: str,
    now: datetime.datetime,
) -> str:
    """Claims the run unless it's done or another attempt holds the lease."""
    snapshot = run_ref.get(transaction=transaction)
    if snapshot.exists:
        run = snapshot.to_dict()
        if run["status"] == "completed":
            return "completed"
        if run["lease_expires"] > now:
            return "in_progress"
    transaction.set(
        run_ref,
        {
            "status": "in_progress",
            "attempt_id": attempt_id,
            "lease_expires": now + LEASE,
            "expire_at": now + RETENTION,
        },
    )
    return "claimed"


@firestore.transactional
def finish_run(
    transaction: firestore.Transaction,
    run_ref: firestore.DocumentReference,
    attempt_id: str,
    succeeded: bool,
) -> None:
    """Completes or releases the run, if this attempt still holds the claim."""
    snapshot = run_ref.get(transaction=transaction)
    if not snapshot.exists or snapshot.to_dict().get("attempt_id") != attempt_id:
        return  # A retry claimed the run after this attempt's lease expired.
    if succeeded:
        transaction.update(run_ref, {"status": "completed"})
    else:
        transaction.delete(run_ref)


@functions_framework.http
def handle_scheduled_job(request: flask.Request) -> tuple[str, int]:
    """Runs a job's work once per scheduled run, even if the run is delivered twice."""
    job_name = request.headers.get("X-CloudScheduler-JobName")
    schedule_time = request.headers.get("X-CloudScheduler-ScheduleTime")
    if not job_name or not schedule_time:
        return "error: missing Cloud Scheduler headers", 400

    # Retries keep the same schedule time. A forced run uses the next
    # scheduled time, so that scheduled run is then skipped.
    run_id = hashlib.sha256(f"{job_name}/{schedule_time}".encode()).hexdigest()
    run_ref = db.collection("scheduler_runs").document(run_id)
    now = datetime.datetime.now(datetime.timezone.utc)
    attempt_id = uuid.uuid4().hex

    status = claim_run(db.transaction(), run_ref, attempt_id, now)
    if status == "completed":
        return "Duplicate run skipped", 200
    if status == "in_progress":
        # A non-2xx response keeps retries going in case that attempt fails.
        return "Run in progress", 409

    try:
        run_job_logic()
    except Exception:
        # Releases the claim so that a retry can process the run.
        finish_run(db.transaction(), run_ref, attempt_id, succeeded=False)
        raise
    finish_run(db.transaction(), run_ref, attempt_id, succeeded=True)
    return "Run processed", 200

防止向 Google Cloud API 目标发送重复请求

当 Cloud Scheduler 通过 HTTP 直接调用 Google Cloud API 端点 (*.googleapis.com) 时，目标 API 会忽略 X-CloudScheduler-JobName 和 X-CloudScheduler-ScheduleTime 标头。例如，调用 workflowexecutions.googleapis.com 以启动Workflows 执行、调用 run.googleapis.com 以运行 Cloud Run 作业，或调用 cloudbuild.googleapis.com 以运行 Cloud Build 构建触发器，都会在每次重试时启动新的运行。

为了提高生产可靠性，请在修改状态的最下游服务中强制执行幂等性。仅在上游层对请求进行去重会缩小重试窗口，但如果下游请求成功且 HTTP 响应在网络中丢失，则上游调用方无法确定操作是否已完成，并且可能仍会重试请求。

为防止在调度 Google Cloud API 目标时重复执行，请结合使用以下方法来实现端到端幂等性：

  • 通过幂等代理路由请求：不要直接从 Cloud Scheduler 调用 *.googleapis.com 端点，而是部署您自己的 Cloud Run 服务或 Cloud Run functions 代理，并配置 Cloud Scheduler 以调用该代理。在代理中，实现本文档中在 HTTP 处理程序中对请求进行去重部分所述的 Python 和 Firestore 原子事务性声明模式。

    代理会读取 X-CloudScheduler-JobName 和 X-CloudScheduler-ScheduleTime，对重试的请求进行去重，并将去重键作为幂等性令牌在请求载荷中传递（例如，在启动 Workflows 执行时在 argument 字段中传递）。

    要求在代理上进行身份验证（例如，使用 --no-allow-unauthenticated 进行部署），并将 Cloud Scheduler 作业配置为使用 OIDC 令牌进行身份验证。在专用服务账号下运行代理，该账号仅具有最低要求的 Identity and Access Management (IAM) 角色（例如目标工作流上的 roles/workflows.invoker 和去重记录的 roles/datastore.user）。

  • 在下游服务中强制执行幂等性：在运行状态更改步骤之前，使用幂等性令牌在数据存储区（例如 Firestore）中执行原子或事务性检查。将令牌传递给服务调用的任何下游 HTTP 端点或 Cloud Run functions，以便这些目标也可以对重试进行去重。如需查看使用 Workflows、Firestore 事务和 Cloud Run functions 的端到端示例，请参阅 Google Cloud博文使用 Workflows 时仅进行一次 API 调用。