Aufgabenhubs

Ein Aufgabenhub ist eine Darstellung des aktuellen Zustands der Anwendung im Speicher, einschließlich aller ausstehenden Arbeiten. Während eine Anwendung ausgeführt wird, speichert der Aufgabenhub kontinuierlich den Fortschritt von Orchestrierungs-, Aktivitäts- und Entitätsfunktionen. Dieser Ansatz stellt sicher, dass die Anwendung die Verarbeitung fortsetzen kann, wo sie unterbrochen wurde, wenn sie nach dem vorübergehenden Anhalten oder Unterbrechen neu gestartet wird. Ein Aufgabenhub ermöglicht es Anwendungen auch, die Compute-Worker dynamisch zu skalieren.

In diesem Artikel wird erläutert, was ein Aufgabenhub speichert, wie Sie Aufgabenhubs konfigurieren und benennen, wie Sie sie mit verschiedenen Speicher-Back-Ends erstellen und verwalten und wie Aufgabenhubs intern funktionieren.

Konzeptionell speichert ein Aufgabenhub die folgenden Informationen:

• Die Instanzzustände aller Orchestrierungs- und Entitätsinstanzen.
• Die zu verarbeitenden Nachrichten, einschließlich:
  • Alle Aktivitätsmeldungen, die Aktivitäten darstellen, die auf die Ausführung warten.
  • Alle Instanznachrichten , die auf die Zustellung an Instanzen warten.

Aktivitätsmeldungen sind zustandslos und können überall verarbeitet werden. Instanznachrichten müssen an eine bestimmte zustandsbehaftete Instanz (Orchestrierung oder Entität) übermittelt werden, die durch ihre Instanz-ID identifiziert wird.

{
  "version": "2.0",
  "extensions": {
    "durableTask": {
      "hubName": "MyTaskHub"
    }
  }
}

{
  "durableTask": {
    "hubName": "MyTaskHub"
  }
}

{
  "version": "2.0",
  "extensions": {
    "durableTask": {
      "hubName": "%MyTaskHub%"
    }
  }
}

{
  "durableTask": {
    "hubName": "%MyTaskHub%"
  }
}

{
  "IsEncrypted": false,
  "Values": {
    "MyTaskHub" : "samplehubname"
  }
}

[FunctionName("HttpStart")]
public static async Task<HttpResponseMessage> Run(
    [HttpTrigger(AuthorizationLevel.Function, methods: "post", Route = "orchestrators/{functionName}")] HttpRequestMessage req,
    [DurableClient(TaskHub = "%MyTaskHub%")] IDurableOrchestrationClient starter,
    string functionName,
    ILogger log)
{
    // Function input comes from the request content.
    object eventData = await req.Content.ReadAsAsync<object>();
    string instanceId = await starter.StartNewAsync(functionName, eventData);

    log.LogInformation($"Started orchestration with ID = '{instanceId}'.");

    return starter.CreateCheckStatusResponse(req, instanceId);
}

{
    "name": "input",
    "taskHub": "%MyTaskHub%",
    "type": "durableClient",
    "direction": "in"
}

{
    "name": "input",
    "taskHub": "%MyTaskHub%",
    "type": "durableClient",
    "direction": "in"
}

{
    "name": "input",
    "taskHub": "%MyTaskHub%",
    "type": "durableClient",
    "direction": "in"
}

@FunctionName("HttpStart")
public HttpResponseMessage httpStart(
        @HttpTrigger(name = "req", route = "orchestrators/{functionName}") HttpRequestMessage<?> req,
        @DurableClientInput(name = "durableContext", taskHub = "%MyTaskHub%") DurableClientContext durableContext,
        @BindingName("functionName") String functionName,
        final ExecutionContext context) {

    DurableTaskClient client = durableContext.getClient();
    String instanceId = client.scheduleNewOrchestrationInstance(functionName);
    context.getLogger().info("Created new Java orchestration with instance ID = " + instanceId);
    return durableContext.createCheckStatusResponse(req, instanceId);
}

Verwenden mehrerer Apps mit separaten Aufgabenhubs

Jede Anwendung, die ein Back-End teilt, sollte eine Verbindung mit einem eigenen Aufgabenhub herstellen, um Konflikte zu vermeiden. Wenn mehrere Apps denselben Aufgabenhub verwenden, konkurrieren sie um Nachrichten, was zu nicht definierten Verhaltensweisen führen kann, einschließlich Orchestrierungen, die unerwartet hängen bleiben. Ein einzelnes Back-End kann mehrere Task-Hubs enthalten; konfigurieren Sie jede Anwendung mit ihrem eigenen Task-Hub.

Verwaltung des Aufgabenhubs für Durable Task Scheduler

In diesem Abschnitt wird beschrieben, wie Sie Aufgabenhubs erstellen und verwalten, wenn Sie das Back-End für dauerhafte Aufgabenplanung verwenden. Erstellen Sie die Ressourcen für den Zeitplan und den Aufgabenhub explizit, bevor Ihre Anwendung sie verwendet.

Erstellen eines Planers und eines Aufgabenhubs

Erstellen Sie einen Planer und einen Aufgabenhub mithilfe des Azure Portals, Azure CLI, Azure Resource Manager (ARM) oder Bicep.

@description('The name of the Durable Task Scheduler.')
param schedulerName string

@description('The name of the task hub.')
param taskHubName string

@description('The location for all resources.')
param location string = resourceGroup().location

resource scheduler 'Microsoft.DurableTask/schedulers@2025-04-01-preview' = {
  name: schedulerName
  location: location
  properties: {
    ipAllowlist: [
      '0.0.0.0/0'
    ]
    sku: {
      name: 'Dedicated'
      capacity: 1
    }
  }
}

resource taskHub 'Microsoft.DurableTask/schedulers/taskHubs@2025-04-01-preview' = {
  parent: scheduler
  name: taskHubName
  properties: {}
}

In den vorherigen Beispielen wird die dedizierte SKU verwendet. Der Durable Task Scheduler bietet auch eine Consumption SKU. Weitere Informationen zum Verwalten von Ressourcen im Durable Task Scheduler finden Sie unter Develop with Durable Task Scheduler.

Konfigurieren der identitätsbasierten Authentifizierung

Der dauerhafte Aufgabenplaner unterstützt nur die verwaltete Identitätsauthentifizierung. Es unterstützt keine Verbindungszeichenfolgen mit Speicherschlüsseln. Weisen Sie der verwalteten Identität die entsprechende rollenbasierte Zugriffssteuerungsrolle (RBAC) zu, und konfigurieren Sie Ihre App so, dass diese Identität verwendet wird.

Die folgenden Rollen stehen zur Verfügung:

Verwenden Sie nach Möglichkeit vom Benutzer zugewiesene verwaltete Identitäten, da sie nicht an den Lebenszyklus der App gebunden sind und sie nach dem Entfernen der App wiederverwenden können.

PUT https://management.azure.com/subscriptions/SUBSCRIPTION_ID/resourceGroups/RESOURCE_GROUP/providers/Microsoft.DurableTask/schedulers/SCHEDULER_NAME/providers/Microsoft.Authorization/roleAssignments/ROLE_ASSIGNMENT_ID?api-version=2022-04-01

{
  "properties": {
    "roleDefinitionId": "/subscriptions/SUBSCRIPTION_ID/providers/Microsoft.Authorization/roleDefinitions/ROLE_DEFINITION_ID",
    "principalId": "MANAGED_IDENTITY_PRINCIPAL_ID",
    "principalType": "ServicePrincipal"
  }
}

@description('The principal ID of the managed identity.')
param identityPrincipalId string

@description('The Durable Task Data Contributor role definition ID.')
var durableTaskDataContributorRoleId = subscriptionResourceId(
  'Microsoft.Authorization/roleDefinitions',
  'ROLE_DEFINITION_ID'
)

resource roleAssignment 'Microsoft.Authorization/roleAssignments@2022-04-01' = {
  name: guid(scheduler.id, identityPrincipalId, durableTaskDataContributorRoleId)
  scope: scheduler
  properties: {
    roleDefinitionId: durableTaskDataContributorRoleId
    principalId: identityPrincipalId
    principalType: 'ServicePrincipal'
  }
}

Nachdem Sie die Identität zugewiesen haben, fügen Sie Ihrer App die folgenden Umgebungsvariablen hinzu:

Vollständige Identitätskonfigurationsdetails finden Sie unter Configure managed identity for Durable Task Scheduler.

Arbeitsaufgaben

Die Aktivitätsnachrichten und Instanzmeldungen im Aufgabenhub stellen die Arbeit dar, die die Anwendung verarbeiten muss. Während die Anwendung ausgeführt wird, ruft sie kontinuierlich Arbeitsaufgaben vom Aufgabenhub ab. Jede Arbeitsaufgabe verarbeitet eine oder mehrere Nachrichten. Es gibt zwei Arten von Arbeitsaufgaben:

• Aktivitätsarbeitselemente: Führen Sie eine Aktivitätsfunktion aus, um eine Aktivitätsnachricht zu verarbeiten.
• Orchestrator-Arbeitsaufgaben: Führen Sie eine Orchestrator- oder Entitätsfunktion aus, um eine oder mehrere Instanznachrichten zu verarbeiten.

Mitarbeiter können mehrere Arbeitsaufgaben gleichzeitig verarbeiten, vorbehaltlich der konfigurierten Parallelitätsgrenzwerte pro Mitarbeiter.

Sobald ein Worker ein Arbeitselement abgeschlossen hat, committet er die Effekte zurück an den Aufgabenhub. Diese Effekte variieren je nach Art der ausgeführten Funktion:

• Eine abgeschlossene Aktivitätsfunktion erstellt eine Instanznachricht, die das Ergebnis enthält, das an die übergeordnete Orchestratorinstanz adressiert ist.
• Eine abgeschlossene Orchestratorfunktion aktualisiert den Orchesterstatus und den Verlauf und kann neue Nachrichten erstellen.
• Eine abgeschlossene Entitätsfunktion aktualisiert den Entitätsstatus und kann auch neue Instanzmeldungen erstellen.

Bei Orchestrierungen stellt jedes Arbeitselement eine Episode der Ausführung dieser Orchestrierung dar. Eine Episode ist eine Runde des Orchestrators, der ausgeführt wird, die verfügbaren Ergebnisse verarbeitet und dann anhält, bis das nächste Ergebnis eingeht. Beispielsweise beginnt eine Episode, wenn die Orchestrierung beginnt, wenn eine Aktivität abgeschlossen ist und ein Ergebnis zurückgibt, oder wenn ein externes Ereignis eingeht. Die Episode endet, wenn der Orchestrator beendet oder einen Punkt erreicht, an dem er auf neue Nachrichten warten muss.

Ausführungsbeispiel

Betrachten Sie eine Fan-Out-Fan-In-Orchestrierung, bei der zwei Aktivitäten parallel gestartet werden und gewartet wird, bis beide abgeschlossen wurden:

[FunctionName("Example")]
public static async Task Run([OrchestrationTrigger] IDurableOrchestrationContext context)
{
    Task t1 = context.CallActivityAsync<int>("MyActivity", 1);
    Task t2 = context.CallActivityAsync<int>("MyActivity", 2);
    await Task.WhenAll(t1, t2);
}

module.exports = df.orchestrator(function*(context){
    const tasks = [];
    tasks.push(context.df.callActivity("MyActivity", 1));
    tasks.push(context.df.callActivity("MyActivity", 2));
    yield context.df.Task.all(tasks);
});

def orchestrator_function(context: df.DurableOrchestrationContext):
    tasks = []
    tasks.append(context.call_activity("MyActivity", 1))
    tasks.append(context.call_activity("MyActivity", 2))
    yield context.task_all(tasks)

param($Context)

$Tasks = @()

$Tasks += Invoke-DurableActivity -FunctionName 'MyActivity' -Input 1 -NoWait
$Tasks += Invoke-DurableActivity -FunctionName 'MyActivity' -Input 2 -NoWait

Wait-DurableTask -Task $Tasks

@FunctionName("Example")
public void exampleOrchestrator(
        @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx) {
    Task<Void> t1 = ctx.callActivity("MyActivity", 1);
    Task<Void> t2 = ctx.callActivity("MyActivity", 2);
    ctx.allOf(List.of(t1, t2)).await();
}

using Microsoft.DurableTask;

public class Example : TaskOrchestrator<object?, object?>
{
    public override async Task<object?> RunAsync(TaskOrchestrationContext context, object? input)
    {
        Task t1 = context.CallActivityAsync("MyActivity", 1);
        Task t2 = context.CallActivityAsync("MyActivity", 2);
        await Task.WhenAll(t1, t2);
        return null;
    }
}

from durabletask import task

def example_orchestrator(ctx: task.OrchestrationContext, _):
    t1 = ctx.call_activity("my_activity", input=1)
    t2 = ctx.call_activity("my_activity", input=2)
    yield task.when_all([t1, t2])

import com.microsoft.durabletask.TaskOrchestration;
import com.microsoft.durabletask.TaskOrchestrationContext;

public class Example implements TaskOrchestration {
    @Override
    public void run(TaskOrchestrationContext ctx) {
        Task<Void> t1 = ctx.callActivity("MyActivity", 1, Void.class);
        Task<Void> t2 = ctx.callActivity("MyActivity", 2, Void.class);
        ctx.allOf(List.of(t1, t2)).await();
    }
}

Nachdem diese Orchestrierung von einem Client initiiert wurde, verarbeitet die Anwendung sie als Abfolge von Arbeitsaufgaben. Jedes abgeschlossene Arbeitselement aktualisiert beim Committen den Status des Aufgabenhubs. Hier sind die Schritte aufgeführt:

1. Ein Client fordert an, eine neue Orchestrierung mit der Instanz-ID "123" zu starten. Nachdem der Client diese Anforderung abgeschlossen hat, enthält der Aufgabenhub einen Platzhalter für den Orchestrierungsstatus und eine Instanzmeldung:

   

   Die Bezeichnung ExecutionStarted ist eine von vielen history-Ereignistypen die die verschiedenen Arten von Nachrichten und Ereignissen identifizieren, die an der Geschichte einer Orchestrierung teilnehmen.

2. Ein Worker führt einen Orchestrator-Arbeitsvorgang aus, um die ExecutionStarted Nachricht zu verarbeiten. Sie ruft die Orchestratorfunktion auf, die mit der Ausführung des Orchestrierungscodes beginnt. Dieser Code plant zwei Aktivitäten und beendet die Ausführung, wenn sie auf die Ergebnisse wartet.

   

   Der Laufzeitzustand ist jetzt Running, und der Verlauf zeichnet diese erste Episode auf: Der Orchestrator hat begonnen, die Ausführung hat begonnen, zwei Aufgaben wurden angesetzt, und der Orchestrator hat die Episode abgeschlossen.

3. Ein Worker führt ein Aktivitätsarbeitselement aus, um eine der TaskScheduled-Nachrichten zu verarbeiten. Sie ruft die Aktivitätsfunktion mit der Eingabe "2" auf. Wenn die Aktivitätsfunktion abgeschlossen ist, wird eine TaskCompleted Nachricht erstellt, die das Ergebnis enthält.

   

4. Ein Worker führt einen Orchestrator-Arbeitsvorgang aus, um die TaskCompleted Nachricht zu verarbeiten. Wenn die Orchestrierung noch im Arbeitsspeicher zwischengespeichert ist, kann sie einfach die Ausführung fortsetzen. Andernfalls gibt der Worker zuerst den Verlauf wieder, um den aktuellen Zustand der Orchestrierung wiederherzustellen. Dann setzt sie die Orchestrierung fort und liefert das Ergebnis der Aktivität. Nach dem Empfang dieses Ergebnisses wartet die Orchestrierung weiterhin auf das Ergebnis der anderen Aktivität, sodass die Ausführung erneut beendet wird.

   

   Das Protokoll zeichnet die zweite Episode auf: Die Aufgabe wurde abgeschlossen, und der Orchestrator wurde erneut angehalten.

5. Ein Worker führt ein Aktivitätsarbeitselement aus, um die verbleibende TaskScheduled-Nachricht zu verarbeiten. Sie ruft die Aktivitätsfunktion mit der Eingabe "1" auf.

   

6. Ein Worker führt ein weiteres Orchestratorarbeitselement aus, um die TaskCompleted-Nachricht zu verarbeiten. Nach Erhalt dieses zweiten Ergebnisses wird die Orchestrierung abgeschlossen.

   

   Der Laufzeitstatus ist jetzt Completed, und der Verlauf zeichnet den dritten und letzten Abschnitt auf: die zweite Aufgabe abgeschlossen und die Ausführung beendet.

Darstellung im Speicher

Anbieter für dauerhafte Aufgabenplanung

Der Durable Task Scheduler ist ein vollständig verwalteter Back-End-Anbieter, der den gesamten Task Hub-Zustand intern speichert. Im Gegensatz zu den Bring-your-own (BYO)-Speicheranbietern müssen Sie keine zugrunde liegende Speicherinfrastruktur einrichten oder verwalten. Jede Scheduler-Ressource (Microsoft.DurableTask/schedulers) verfügt über dedizierte Compute- und Memory-Ressourcen und kann einen oder mehrere Task Hubs (Microsoft.DurableTask/schedulers/taskHubs) enthalten.

Da der permanente Aufgabenplaner den Speicher intern verwaltet, können Sie die zugrunde liegenden Daten nicht direkt prüfen. Verwenden Sie stattdessen das Dashboard "Durable Task Scheduler", um Orchestrierungsinstanzen zu überwachen und abzufragen.