Hubs de tâches

Un hub de tâches est une représentation de l’état actuel de l’application dans le stockage, y compris tout le travail en attente. Pendant qu’une application s’exécute, le hub de tâches stocke continuellement la progression de l’orchestration, de l’activité et des fonctions d’entité. Cette approche garantit que l’application peut reprendre le traitement là où elle s’est arrêtée si elle redémarre après avoir été temporairement arrêtée ou interrompue. Un hub de tâches permet également aux applications de faire évoluer dynamiquement les travailleurs informatiques.

Cet article explique ce qu’un hub de tâches stocke, comment configurer et nommer des hubs de tâches, comment les créer et les gérer avec différents back-ends de stockage et comment fonctionnent en interne les hubs de tâches.

Conceptuellement, un hub de tâches stocke les informations suivantes :

• Les états d’instance de toutes les instances d’orchestration et d’entité.
• Messages à traiter, notamment :
  • Tous les messages d’activité qui représentent les activités en attente d’exécution.
  • Tous les messages d’instance qui attendent d’être remis aux instances.

Les messages d’activité sont sans état et peuvent être traités n’importe où. Les messages d’instance doivent être remis à une instance avec état particulier (orchestration ou entité), identifiée par son ID d’instance.

{
  "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);
}

Utiliser plusieurs applications avec des hubs de tâches distincts

Chaque application qui partage un back-end doit se connecter à son propre hub de tâches pour éviter les conflits. Si plusieurs applications utilisent le même hub de tâches, elles se concurrencent pour les messages, ce qui peut entraîner un comportement non défini, y compris des orchestrations bloquées de manière inattendue. Un serveur principal unique peut contenir plusieurs hubs de tâches ; configurez chaque application avec sa propre application.

Gestion durable du hub de tâches du planificateur de tâches

Cette section explique comment créer et gérer des hubs de tâches lors de l’utilisation du serveur principal du planificateur de tâches durable. Créez explicitement les ressources du planificateur et du hub de tâches avant que votre application ne les utilise.

Créer un planificateur et un hub de tâches

Créez un planificateur et un hub de tâches à l’aide du portail Azure, Azure CLI, Azure Resource Manager (ARM) ou 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: {}
}

Les exemples précédents utilisent la référence SKU dédiée. Le planificateur de tâches durables propose également une référence de consommation. Pour plus d’informations sur la gestion des ressources du Planificateur de Tâches Durables, consultez Développer avec le Planificateur de Tâches Durables.

Configurer l’authentification basée sur l’identité

Durable Task Scheduler prend uniquement en charge l’authentification d’identité managée. Il ne prend pas en charge les chaînes de connexion avec des clés de stockage. Attribuez le rôle de contrôle d’accès en fonction du rôle (RBAC) approprié à une identité managée et configurez votre application pour utiliser cette identité.

Les rôles suivants sont disponibles :

Utilisez les identités managées affectées par l’utilisateur si possible, car elles ne sont pas liées au cycle de vie de l’application et vous pouvez les réutiliser une fois l’application supprimée.

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

Après avoir affecté l’identité, ajoutez les variables d’environnement suivantes à votre application :

Pour des détails complets sur la configuration de l’identité, consultez Configurer l’identité managée pour Durable Task Scheduler.

Éléments de travail

Les messages d’activité et les messages d’instance dans le hub de tâches représentent le travail que l’application doit traiter. Pendant que l’application est en cours d’exécution, elle extrait en continu des éléments de travail à partir du hub de tâches. Chaque élément de travail traite un ou plusieurs messages. Il existe deux types d’éléments de travail :

• Éléments de travail liés à l'activité : exécutez une fonction d'activité pour traiter un message relatif à l'activité.
• Éléments de travail Orchestrator : exécutez une fonction d’orchestrateur ou d’entité pour traiter un ou plusieurs messages d’instance.

Les travailleurs peuvent traiter plusieurs éléments de travail en même temps, sous réserve des limites de concurrence configurées par travailleur.

Une fois qu’un worker a terminé un élément de travail, il valide les effets dans le hub de tâches. Ces effets varient selon le type de fonction exécuté :

• Une fonction d’activité terminée crée un message d’instance contenant le résultat, adressé à l’instance d’orchestrateur parent.
• Une fonction d’orchestrateur terminée met à jour l’état et l’historique de l’orchestration, et peut créer de nouveaux messages.
• Une fonction d’entité terminée met à jour l’état de l’entité et peut également créer des messages d’instance.

Pour les orchestrations, chaque élément de travail représente un épisode de l’exécution de cette orchestration. Un épisode correspond à un cycle complet de l'orchestrateur, traitant les résultats disponibles, puis s'interrompant jusqu'à l'arrivée du prochain résultat. Par exemple, un épisode démarre lorsque l’orchestration commence, lorsqu’une activité se termine et retourne un résultat, ou lorsqu’un événement externe arrive. L'épisode se termine lorsque l'orchestrateur achève ses opérations ou atteint un point où il doit attendre de nouveaux messages.

Exemple d’exécution

Considérez une orchestration Fan-out-Fan-in qui démarre deux activités en parallèle et attend que les deux se terminent :

[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();
    }
}

Une fois cette orchestration lancée par un client, l’application la traite comme une séquence d’éléments de travail. Chaque élément de travail terminé met à jour l’état du hub de tâches lors de sa validation. Ces étapes sont les suivantes :

1. Un client demande de démarrer une nouvelle orchestration avec l’ID d’instance « 123 ». Une fois que le client a terminé cette requête, le hub de tâches contient un espace réservé pour l’état d’orchestration et un message d’instance :

   

   L'étiquette ExecutionStarted est l'un des nombreux types d'événements history qui identifient les différents types de messages et d'événements participant à l'historique d'une orchestration.

2. Un opérateur exécute un élément de travail d’orchestrateur pour traiter le ExecutionStarted message. Il appelle la fonction d’orchestrateur qui commence à exécuter le code d’orchestration. Ce code planifie deux activités, puis cesse de s’exécuter lorsqu’il attend les résultats.

   

   L’état d’exécution est maintenant Running, et l’historique enregistre ce premier épisode : l’orchestrateur a démarré, l’exécution démarrée, deux tâches ont été planifiées et l’orchestrateur a terminé l’épisode.

3. Un opérateur exécute un élément de travail lié à une activité pour traiter l'un des TaskScheduled messages. Elle appelle la fonction d’activité avec l’entrée « 2 ». Une fois la fonction d’activité terminée, elle crée un TaskCompleted message contenant le résultat.

   

4. Un opérateur exécute un élément de travail d’orchestrateur pour traiter le TaskCompleted message. Si l’orchestration est toujours mise en cache en mémoire, elle peut simplement reprendre l’exécution. Sinon, le worker relit d’abord l’historique pour récupérer l’état actuel de l’orchestration. Ensuite, il poursuit l’orchestration, fournissant le résultat de l’activité. Après avoir reçu ce résultat, l'orchestration attend toujours le résultat de l'autre activité, elle s'arrête donc à nouveau.

   

   L’historique enregistre le deuxième épisode : la tâche s’est terminée et l’orchestrateur a suspendu à nouveau.

5. Un worker exécute un élément de travail d’activité pour traiter le message TaskScheduled restant. Il appelle la fonction d’activité avec l’entrée « 1 ».

   

6. Un travailleur exécute un autre élément de travail d’orchestrateur pour traiter le TaskCompleted message. Après avoir reçu ce deuxième résultat, l’orchestration se termine.

   

   L’état d’exécution est maintenant Completed, et l’historique enregistre le troisième et dernier épisode : la deuxième tâche est terminée et l’exécution terminée.

Représentation dans le stockage

Fournisseur planificateur de tâches durables

Le Durable Task Scheduler est un fournisseur backend entièrement géré qui stocke l’état de tous les hubs de tâches en interne. Contrairement aux fournisseurs de stockage BYO (bring-your-own), vous n’avez pas besoin de configurer ou de gérer une infrastructure de stockage sous-jacente. Chaque ressource du planificateur (Microsoft.DurableTask/schedulers) possède des ressources de calcul et de mémoire dédiées, et peut contenir un ou plusieurs hubs de tâches (Microsoft.DurableTask/schedulers/taskHubs).

Étant donné que durable Task Scheduler gère le stockage en interne, vous ne pouvez pas inspecter directement les données sous-jacentes. Utilisez plutôt le tableau de bord Du planificateur de tâches durables pour surveiller et interroger des instances d’orchestration.