Centri operativi

Un hub attività è una rappresentazione dello stato corrente dell'applicazione nell'archiviazione, incluso tutto il lavoro in sospeso. Durante l'esecuzione di un'applicazione, l'hub delle attività salva continuamente il progresso delle funzioni di orchestrazione, attività ed entità. Questo approccio garantisce che l'applicazione possa riprendere l'elaborazione in cui è stata interrotta se viene riavviata dopo essere stata temporaneamente arrestata o interrotta. Un hub delle attività consente alle applicazioni di ridimensionare dinamicamente le unità di calcolo.

Concettualmente, un hub delle attività archivia le informazioni seguenti:

• Stati delle istanze di tutte le istanze di orchestrazione ed entità.
• Messaggi da elaborare, tra cui:
  • Eventuali messaggi di attività che rappresentano attività in attesa di esecuzione.
  • Tutti i messaggi di istanza in attesa di essere recapitati alle istanze.

I messaggi di attività sono senza stato e possono essere elaborati ovunque. I messaggi di istanza devono essere recapitati a una specifica istanza con stato (orchestrazione o entità), identificata dall'ID istanza.

Elementi di lavoro

I messaggi di attività e i messaggi di istanza nell'hub attività rappresentano il lavoro che l'applicazione deve elaborare. Mentre l'applicazione è in esecuzione, recupera continuamente elementi di lavoro dall'hub attività. Ogni elemento di lavoro elabora uno o più messaggi. Distinguiamo due tipi di elementi di lavoro:

• Elementi di lavoro attività: eseguire una funzione di attività per elaborare un messaggio di attività.
• Elemento di lavoro di Orchestrator: eseguire un agente di orchestrazione o una funzione di entità per elaborare uno o più messaggi di istanza.

I lavoratori possono elaborare più elementi di lavoro contemporaneamente, soggetti ai limiti di concorrenza configurati per ogni lavoratore.

Una volta che un lavoratore completa un elemento di lavoro, applica gli effetti nell'hub attività. Questi effetti variano in base al tipo di funzione eseguita:

• Una funzione di attività completata crea un messaggio di istanza contenente il risultato, indirizzato all'istanza orchestratore padre.
• Una funzione dell'agente di orchestrazione completata aggiorna lo stato e la cronologia dell'orchestrazione e può creare nuovi messaggi.
• Una funzione di entità completata aggiorna lo stato dell'entità e può anche creare nuovi messaggi di istanza.

Per le orchestrazioni, ogni elemento di lavoro rappresenta un episodio dell'esecuzione dell'orchestrazione. Un episodio inizia quando ci sono nuovi messaggi per l'orchestratore da elaborare. Un messaggio di questo tipo può indicare che l'orchestrazione deve iniziare; o può indicare che un'attività, una chiamata di entità, un timer o una suborchestrazione è stata completata; oppure può rappresentare un evento esterno. Il messaggio attiva un elemento di lavoro che consente all'agente di orchestrazione di elaborare il risultato e di continuare con l'episodio successivo. ** L'episodio termina quando l'orchestratore completa oppure raggiunge un punto in cui deve attendere nuovi messaggi.

Esempio di esecuzione

Valutare l'orchestrazione fan-out-fan-in che avvia due attività in parallelo e attende che entrambe vengano completate.

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

Dopo l'avvio di questa orchestrazione da parte di un client, l'applicazione lo elabora come sequenza di elementi di lavoro. Ogni elemento di lavoro completato aggiorna lo stato dell'hub attività quando esegue il commit. Questa è la procedura:

1. Un client richiede di avviare una nuova orchestrazione con id istanza "123". Dopo che il cliente completa la richiesta, il task hub contiene un segnaposto per lo stato di orchestrazione e un messaggio di istanza.

L'etichetta ExecutionStarted è uno dei molti tipi di eventi storici che identificano i vari tipi di messaggi ed eventi coinvolti nella cronologia di un'orchestrazione.

2. Un processo di lavoro esegue un elemento di lavoro dell'agente di orchestrazione per elaborare il messaggio ExecutionStarted. Chiama la funzione di orchestrazione che avvia l'esecuzione del codice di orchestrazione. Questo codice pianifica due attività e quindi arresta l'esecuzione quando è in attesa dei risultati. Quando il lavoratore effettua il commit di questo elemento di lavoro, l'hub delle attività contiene

Lo stato di runtime è ora Running, sono stati aggiunti due nuovi TaskScheduled messaggi e la cronologia contiene ora i cinque eventi OrchestratorStarted, , ExecutionStartedTaskScheduled, TaskScheduled, OrchestratorCompleted. Questi eventi rappresentano il primo episodio dell'esecuzione dell'orchestrazione.

3. Un lavoratore esegue un attività di lavoro per elaborare uno dei TaskScheduled messaggi. Invoca la funzione attività con l'input "2". Al termine della funzione di attività, viene creato un TaskCompleted messaggio contenente il risultato. Quando il lavoratore effettua il commit di questo elemento di lavoro, l'hub delle attività contiene

4. Un lavoratore esegue un elemento di lavoro di orchestrazione per elaborare il TaskCompleted messaggio. Se l'orchestrazione è ancora memorizzata nella cache, può semplicemente riprendere l'esecuzione. Altrimenti, il processo di lavoro ricalca la cronologia per recuperare lo stato corrente dell'orchestrazione. Continua quindi l'orchestrazione, fornendo il risultato dell'attività. Dopo aver ricevuto questo risultato, l'orchestrazione è ancora in attesa del risultato dell'altra attività, quindi ancora una volta interrompe l'esecuzione. Quando il lavoratore effettua il commit di questo elemento di lavoro, l'hub delle attività contiene

La cronologia dell'orchestrazione contiene ora altri tre eventi OrchestratorStarted, , TaskCompletedOrchestratorCompleted. Questi eventi rappresentano il secondo episodio dell'esecuzione dell'orchestrazione.

5. Un processo di lavoro esegue un elemento di lavoro dell'attività per elaborare il messaggio rimanente TaskScheduled. Chiama la funzione attività con l'input "1". Quando il lavoratore effettua il commit di questo elemento di lavoro, l'hub delle attività contiene

6. Un lavoratore esegue un altro elemento di lavoro dell'orchestratore per elaborare il TaskCompleted messaggio. Dopo aver ricevuto questo secondo risultato, l'orchestrazione viene completata. Quando il lavoratore effettua il commit di questo elemento di lavoro, l'hub delle attività contiene

Lo stato di runtime è ora Completede la cronologia dell'orchestrazione ora contiene quattro altri eventiOrchestratorStarted, , TaskCompleted, ExecutionCompletedOrchestratorCompleted. Questi eventi rappresentano il terzo e ultimo episodio dell'esecuzione di questa orchestrazione.

La cronologia finale per l'esecuzione di questa orchestrazione contiene quindi i 12 eventi: OrchestratorStarted, ExecutionStarted, TaskScheduled, TaskScheduled, OrchestratorCompleted, OrchestratorStarted, TaskCompleted, OrchestratorCompleted, OrchestratorStarted, TaskCompleted, ExecutionCompleted, OrchestratorCompleted.

Gestione dell'hub attività Pianificatore di attività durevole

Questa sezione illustra come creare e gestire hub attività quando si usa il back-end dell'Utilità di pianificazione attività permanenti. Creare le risorse del scheduler e del task hub in modo esplicito prima che vengano usate dall'applicazione.

Creare un pianificatore e un hub attività

Creare un pianificatore e un hub di attività usando il portale di Azure, Azure CLI, Azure Resource Manager (ARM) o 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: {}
}

Gli esempi precedenti usano lo SKU dedicato. Il Pianificatore di attività durevole offre anche uno SKU A consumo. Per ulteriori informazioni sulla gestione delle risorse di Durable Task Scheduler, vedere Sviluppare con Durable Task Scheduler.

Configurare l'autenticazione basata su identità

Durable Task Scheduler supporta solo l'autenticazione dell'identità gestita. Non supporta le stringhe di connessione con le chiavi di archiviazione. Assegnare il ruolo con controllo degli accessi in base al ruolo appropriato a un'identità gestita e configurare l'app per l'uso di tale identità.

Sono disponibili i ruoli seguenti:

Usare le identità gestite assegnate dall'utente quando possibile perché non sono associate al ciclo di vita dell'app ed è possibile riutilizzarle dopo la rimozione dell'app.

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

Dopo aver assegnato l'identità, aggiungere le variabili di ambiente seguenti all'app:

Per informazioni complete sulla configurazione delle identità, vedere Configurare l'identità gestita per Durable Task Scheduler.

Più applicazioni

Se più applicazioni condividono lo stesso scheduler, configurare ogni applicazione con un hub delle attività separato. Un unico scheduler può contenere più hub di attività. Ogni applicazione deve connettersi al proprio hub attività per evitare conflitti. In caso contrario, le applicazioni competono per i messaggi, che potrebbero comportare un comportamento indefinito, incluse le orchestrazioni che si bloccano in modo imprevisto.

Rappresentazione nell'archiviazione

Provider del Pianificatore di attività durevoli

Durable Task Scheduler è un provider backend completamente gestito che archivia internamente tutti gli stati dell'hub delle attività. A differenza dei provider di archiviazione BYO (Bring Your Own), non è necessario configurare o gestire un'infrastruttura di archiviazione sottostante. Ogni risorsa dello schedulatore (Microsoft.DurableTask/schedulers) ha risorse di calcolo e memoria dedicate e può contenere uno o più hub di attività (Microsoft.DurableTask/schedulers/taskHubs).

Poiché l'Utilità di pianificazione attività permanenti gestisce internamente l'archiviazione, non è possibile esaminare direttamente i dati sottostanti. In alternativa, utilizzare il dashboard del Pianificatore di attività durevole per monitorare ed eseguire query sulle istanze di orchestrazione.

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

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

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

{
  "version": "2.0",
  "extensions": {
    "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);
}