如何在GCP Workflow中异步执行Cloud Run Job不阻塞流程

方案1：Workflow异步调用Cloud Run Job的Run API #

在Workflow中调用googleapis.run.v1.namespaces.jobs.run时，添加async: true参数，Workflow会发起调用后立即继续后续流程，不会等待Job执行完成。

修改后的代码示例：

- execute_monitor:
    call: googleapis.run.v1.namespaces.jobs.run
    args:
      name: $${"namespaces/" + gcp_project_id + "/jobs/job-name-" + env_name}
      location: $${gcp_region}
      body: {}
    async: true  # 关键参数：异步调用，不阻塞流程

注意：异步调用后无法直接获取Job执行结果，若需追踪状态，后续可通过googleapis.run.v1.namespaces.jobs.executions.list等API查询。

方案2：用Pub/Sub触发Cloud Run Job #

1. 先给Cloud Run Job配置Pub/Sub触发器：在GCP控制台的Cloud Run Job页面，添加触发器选择Pub/Sub并指定对应主题。
2. 在Workflow中调用Pub/Sub的publish API发送消息，触发Job执行。Workflow发送消息后直接继续流程，无需等待Job完成。

Workflow代码示例：

- trigger_job_via_pubsub:
    call: googleapis.pubsub.v1.projects.topics.publish
    args:
      topic: $${"projects/" + gcp_project_id + "/topics/your-topic-name"}
      body:
        messages:
          - data: $${base64.encode(json.encode({"key": "value"}))}  # 可选：传递触发参数

这种方式解耦性更强，还能利用Pub/Sub的重试、死信队列等特性。

方案3：动态创建Job并异步启动 #

如果需要动态生成Job再触发，可先调用jobs.create创建Job，接着异步调用jobs.run启动它：

- create_and_run_job:
    try:
      steps:
        - create_job:
            call: googleapis.run.v1.namespaces.jobs.create
            args:
              parent: $${"namespaces/" + gcp_project_id}
              location: $${gcp_region}
              body:
                apiVersion: "run.googleapis.com/v1"
                kind: "Job"
                metadata:
                  name: $${"monitor-" + uuid()}  # 用UUID避免名称重复
                  annotations:
                    run.googleapis.com/launch-stage: BETA
                spec:
                  template:
                    spec:
                      template:
                        spec:
                          containers:
                            - image: $${gcp_region + "-docker.pkg.dev/" + gcp_project_id + "/repo-name/repo-image:" + env_name}
                              env: [...]
                          timeoutSeconds: 604800  # 7天超时
            result: created_job
        - run_job_async:
            call: googleapis.run.v1.namespaces.jobs.run
            args:
              name: $${created_job.metadata.name}
              location: $${gcp_region}
              body: {}
            async: true
        - log_job_started:
            call: sys.log
            args:
              text: $${"Job created and started: " + created_job.metadata.name}
              severity: INFO
    except:
      as: e
      steps:
        - log_error:
            call: sys.log
            args:
              severity: ERROR
              text: $${"Failed to create/run job: " + json.encode_to_string(e)}

关键注意事项 #

• 异步调用不返回执行结果，需监控状态可单独通过API查询jobs.executions资源。
• Cloud Run Job名称需唯一，建议用UUID或时间戳生成，避免创建失败。
• 确保Workflow服务账号拥有对应权限：roles/run.jobsEditor（创建/运行Job）、roles/pubsub.publisher（Pub/Sub方案）。

内容的提问来源于stack exchange，提问作者unacorn