Cloud Tasks-Warteschlange zum Puffern von Workflowausführungen verwenden

In dieser Anleitung erfahren Sie, wie Sie eine Cloud Tasks-Warteschlange erstellen, mit der die Rate der Workflowausführungen reguliert werden kann.

Es gibt eine maximale Anzahl aktiver Workflowausführungen, die gleichzeitig stattfinden können. Wenn dieses Kontingent aufgebraucht ist und Ausführung von Backlogs deaktiviert ist oder das Kontingent für Backlog-Ausführungen erreicht ist, schlagen alle neuen Ausführungen mit dem HTTP-Statuscode 429 Too many requests fehl. Wenn Sie eine Cloud Tasks-Warteschlange aktivieren, um untergeordnete Workflows mit einer von Ihnen definierten Rate auszuführen, können Sie Probleme im Zusammenhang mit Workflows-Kontingenten vermeiden und eine bessere Ausführungsrate erzielen.

Cloud Tasks ist darauf ausgelegt, „mindestens einmal“ zu liefern. Workflows garantiert jedoch nicht, dass doppelte Anfragen von Cloud Tasks genau einmal verarbeitet werden.

Im folgenden Diagramm ruft ein übergeordneter Workflow untergeordnete Workflows auf, die von einer Cloud Tasks-Warteschlange mit einer angewendeten Versandrate reguliert werden.

Übergeordneter Workflow, der Iterationen eines untergeordneten Workflows über die Cloud Tasks-Warteschlange aufruft

Cloud Tasks-Warteschlange erstellen

Erstellen Sie eine Cloud Tasks-Warteschlange, die Sie im übergeordneten Workflow verwenden können und mit der Sie die Rate der Workflow-Ausführungen regulieren können.

Console

  1. Rufen Sie in der Google Cloud -Console die Seite Cloud Tasks auf:

    Zu Cloud Tasks

  2. Klicken Sie auf  Push-Warteschlange erstellen.

  3. Geben Sie den Namen der Warteschlange ein: queue-workflow-child.

  4. Wählen Sie in der Liste Region die Option us-central1 (Iowa) aus.

  5. Klicken Sie auf Erstellen.

gcloud

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

Untergeordneten Workflow erstellen und bereitstellen

Ein untergeordneter Workflow kann Daten von einem übergeordneten Workflow empfangen und verarbeiten. Erstellen und stellen Sie einen untergeordneten Workflow bereit, der Folgendes ausführt:

  • Empfängt iteration als Argument
  • Wartet 10 Sekunden, um eine Verarbeitung zu simulieren

  • Gibt bei erfolgreicher Ausführung einen String zurück

Console

  1. Rufen Sie in der Google Cloud Console die Seite Workflows auf.

    Zur Seite "Workflows"

  2. Klicken Sie auf  Erstellen.

  3. Geben Sie den Namen workflow-child für den neuen Workflow ein.

  4. Wählen Sie in der Liste Region die Option us-central1 (Iowa) aus.

  5. Wählen Sie in der Liste Dienstkonto das Compute Engine-Standarddienstkonto aus.

  6. Klicken Sie auf Weiter.

  7. Geben Sie im Workflow-Editor die folgende Definition für Ihren Workflow ein:

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

  8. Klicken Sie auf Bereitstellen.

gcloud

  1. Erstellen Sie eine Quellcodedatei für Ihren Workflow:

    touch workflow-child.yaml

  2. Öffnen Sie die Quellcodedatei in einem Texteditor und kopieren Sie den folgenden Workflow in die Datei.

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

  3. Stellen Sie den Workflow bereit:

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

Übergeordneten Workflow erstellen und bereitstellen

Im übergeordneten Workflow werden mehrere Zweige des untergeordneten Workflows mit einer for-Schleife ausgeführt.

  1. Kopieren Sie den Quellcode, der den übergeordneten Workflow definiert:

    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"}

    Der Workflow besteht aus den folgenden Teilen:

    • Eine Map, mit der Konstanten zugewiesen werden, die auf den untergeordneten Workflow und den Namen der Cloud Tasks-Warteschlange verweisen. Weitere Informationen finden Sie unter Maps.

    • Eine for-Schleife, die iterativ ausgeführt wird, um den untergeordneten Workflow aufzurufen. Weitere Informationen finden Sie unter Iteration.

    • Ein Workflow-Schritt, mit dem eine große Anzahl von Aufgaben erstellt und der Cloud Tasks-Warteschlange hinzugefügt wird, um den untergeordneten Workflow auszuführen. Weitere Informationen finden Sie unter Cloud Tasks API-Connector.

  2. Stellen Sie den Workflow bereit:

    Console

    1. Rufen Sie in der Google Cloud Console die Seite Workflows auf:

      Zur Seite "Workflows"

    2. Klicken Sie auf  Erstellen.

    3. Geben Sie den Namen workflow-parent für den neuen Workflow ein.

    4. Wählen Sie in der Liste Region die Option us-central1 (Iowa) aus.

    5. Wählen Sie in der Liste Dienstkonto das Compute Engine-Standarddienstkonto aus.

    6. Klicken Sie auf Weiter.

    7. Fügen Sie im Workflow-Editor die Definition für den übergeordneten Workflow ein.

    8. Klicken Sie auf Bereitstellen.

    gcloud

    1. Erstellen Sie eine Quellcodedatei für Ihren Workflow:

      touch workflow-parent.yaml

    2. Öffnen Sie die Quellcodedatei in einem Texteditor und fügen Sie die Definition für den übergeordneten Workflow ein.

    3. Stellen Sie den Workflow bereit:

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

Übergeordneten Workflow ohne Ratenbeschränkungen ausführen

Führen Sie den übergeordneten Workflow aus, um die untergeordneten Workflows über die Cloud Tasks-Warteschlange aufzurufen. Die Ausführungen sollten etwa 10 Sekunden dauern.

Console

  1. Rufen Sie in der Google Cloud Console die Seite Workflows auf:

    Zur Seite "Workflows"

  2. Klicken Sie auf der Seite Workflows auf den Workflow workflow-parent, um die Detailseite aufzurufen.

  3. Klicken Sie auf der Seite Workflow-Details auf  Ausführen.

  4. Klicken Sie noch einmal auf Ausführen.

  5. Kehren Sie während der Ausführung des übergeordneten Workflows zur Seite Workflows zurück und klicken Sie auf den Workflow workflow-child, um die Detailseite aufzurufen.

  6. Klicken Sie auf den Tab Ausführungen.

    Es sollten Ausführungen des untergeordneten Workflows etwa zur gleichen Zeit angezeigt werden, ähnlich wie im Folgenden:

    Details zu den Ausführungen untergeordneter Workflows, die etwa zur gleichen Zeit ausgeführt werden.

gcloud

  1. Führen Sie den Workflow aus:

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

  2. Um zu prüfen, ob eine Workflowausführung ausgelöst wurde, listen Sie die letzten vier Ausführungen auf:

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

    Da die Anzahl der Ausführungen (100) unter dem Gleichzeitigkeitslimit für Workflows liegt, sollten die Ergebnisse in etwa so aussehen: Probleme mit dem Kontingent können auftreten, wenn Sie Tausende von Ausführungen gleichzeitig senden. 

    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:

Sie haben einen Workflow erstellt und bereitgestellt, der 100 Iterationen des untergeordneten Workflows aufruft.

Übergeordneten Workflow mit Ratenbegrenzungen ausführen

Wenden Sie ein Ratenlimit von einem Dispatch pro Sekunde auf die Cloud Tasks-Warteschlange an und führen Sie dann den übergeordneten Workflow aus.

Console

  1. Rufen Sie in der Google Cloud -Console die Seite Cloud Tasks auf:

    Zu Cloud Tasks

  2. Klicken Sie auf queue-workflow-child, die Cloud Tasks-Warteschlange, die Sie erstellt haben, und dann auf Warteschlange bearbeiten.

  3. Geben Sie im Abschnitt Ratenlimits für Aufgabenweiterleitungen im Feld Max. Weiterleitungen 1 ein.

  4. Klicken Sie auf Speichern.

  5. Rufen Sie die Seite Workflows auf:

    Zur Seite "Workflows"

  6. Klicken Sie auf den Workflow workflow-parent, um die zugehörige Detailseite aufzurufen.

  7. Klicken Sie auf der Seite Workflow-Details auf  Ausführen.

  8. Klicken Sie noch einmal auf Ausführen.

  9. Kehren Sie während der Ausführung des übergeordneten Workflows zur Seite Workflows zurück und klicken Sie auf den Workflow workflow-child, um die Detailseite aufzurufen.

  10. Klicken Sie auf den Tab Ausführungen.

    Sie sollten Ausführungen des untergeordneten Workflows sehen, die mit einer Anfrage pro Sekunde ausgeführt werden, ähnlich wie im Folgenden:

    Details zum untergeordneten Workflow, der mit einer Anfrage pro Sekunde ausgeführt wird.

gcloud

  1. Aktualisieren Sie die Cloud Tasks-Warteschlange, um ein Ratenlimit von einer Weiterleitung pro Sekunde anzuwenden:

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

  2. Führen Sie den Workflow aus:

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

  3. Um zu prüfen, ob eine Workflowausführung ausgelöst wurde, listen Sie die letzten vier Ausführungen auf:

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

    Die Ergebnisse sollten in etwa so aussehen, wobei pro Sekunde ein Workflow ausgeführt wird: 

    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:

Sie haben einen Workflow bereitgestellt, der 100 Iterationen des untergeordneten Workflows mit einer Dispatch-Rate von einer Ausführung pro Sekunde aufruft.