Guida introduttiva: Creare un'app con gli SDK di Durable Task e il Durable Task Scheduler

Prerequisiti

Prima di iniziare:

• Assicurarsi di avere .NET 10 SDK o versione successiva.
• Installare Docker per l'esecuzione dell'emulatore.
• Clonare il repository GitHub Durable Task Scheduler per usare l'esempio rapido.

• Assicurarsi di avere Python 3.9+ o versione successiva.
• Installare Docker per l'esecuzione dell'emulatore.
• Clonare il repository GitHub Durable Task Scheduler per usare l'esempio rapido.

• Assicurarsi di avere Java 8 o 11.
• Installare Docker per l'esecuzione dell'emulatore.
• Clonare il repository GitHub Durable Task Scheduler per usare l'esempio rapido.

• Assicurarsi di avere Node.js 22 o versione successiva.
• Installare Docker per l'esecuzione dell'emulatore.
• Clonare il repository GitHub Durable Task Scheduler per usare l'esempio rapido.

• Installa Go 1.25 o versione successiva.
• Installa Docker o un runtime container compatibile.

Configurare l'emulatore del pianificatore di attività durevole

L'emulatore simula il Task Scheduler durevole e un hub di attività all'interno di un contenitore Docker, rendendolo ideale per lo sviluppo locale.

Il codice di esempio si connette all'emulatore per impostazione predefinita quando non vengono impostate variabili di ambiente.

1. Dalla directory radice Azure-Samples/Durable-Task-Scheduler, passare alla directory di esempio .NET SDK.

   cd samples/durable-task-sdks/dotnet/FanOutFanIn
   

2. Eseguire il pull dell'immagine Docker per l'emulatore.

   docker pull mcr.microsoft.com/dts/dts-emulator:latest
   

3. Eseguire l'emulatore. Il contenitore può richiedere alcuni secondi per essere pronto.

   docker run --name dtsemulator -d -p 8080:8080 -p 8082:8082 mcr.microsoft.com/dts/dts-emulator:latest
   

Poiché il codice di esempio usa automaticamente le impostazioni dell'emulatore predefinite, non è necessario impostare variabili di ambiente. Le impostazioni predefinite dell'emulatore per questa guida introduttiva sono:

• Punto finale: http://localhost:8080
• Hub attività: default

1. Dalla directory radice Azure-Samples/Durable-Task-Scheduler passare alla directory di esempio Python SDK.

   cd samples/durable-task-sdks/python/fan-out-fan-in
   

2. Eseguire il pull dell'immagine Docker per l'emulatore.

   docker pull mcr.microsoft.com/dts/dts-emulator:latest
   

3. Eseguire l'emulatore. Il contenitore può richiedere alcuni secondi per essere pronto.

   docker run --name dtsemulator -d -p 8080:8080 -p 8082:8082 mcr.microsoft.com/dts/dts-emulator:latest
   

Poiché il codice di esempio usa automaticamente le impostazioni dell'emulatore predefinite, non è necessario impostare variabili di ambiente. Le impostazioni predefinite dell'emulatore per questa guida introduttiva sono:

• Punto finale: http://localhost:8080
• Hub attività: default

1. Azure-Samples/Durable-Task-Scheduler Dalla directory radice passare alla directory di esempio di Java SDK.

   cd samples/durable-task-sdks/java/fan-out-fan-in
   

2. Eseguire il pull dell'immagine Docker per l'emulatore.

   docker pull mcr.microsoft.com/dts/dts-emulator:latest
   

3. Eseguire l'emulatore. Il contenitore può richiedere alcuni secondi per essere pronto.

   docker run --name dtsemulator -d -p 8080:8080 -p 8082:8082 mcr.microsoft.com/dts/dts-emulator:latest
   

Poiché il codice di esempio usa automaticamente le impostazioni dell'emulatore predefinite, non è necessario impostare variabili di ambiente. Le impostazioni predefinite dell'emulatore per questa guida introduttiva sono:

• Punto finale: http://localhost:8080
• Hub attività: default

1. Dalla directory radice Azure-Samples/Durable-Task-Scheduler passare alla directory di esempio JavaScript SDK.

   cd samples/durable-task-sdks/javascript/fan-out-fan-in
   

2. Eseguire il pull dell'immagine Docker per l'emulatore.

   docker pull mcr.microsoft.com/dts/dts-emulator:latest
   

3. Eseguire l'emulatore. Il contenitore può richiedere alcuni secondi per essere pronto.

   docker run --name dtsemulator -d -p 8080:8080 -p 8082:8082 mcr.microsoft.com/dts/dts-emulator:latest
   

Poiché il codice di esempio usa automaticamente le impostazioni dell'emulatore predefinite, non è necessario impostare variabili di ambiente. Le impostazioni predefinite dell'emulatore per questa guida introduttiva sono:

• Punto finale: http://localhost:8080
• Hub attività: default

1. Avvia un emulatore dedicato. Assegna le porte al loopback per mantenere l'emulatore non autenticato locale sulla tua macchina.

   docker run --name go-dts-emulator -d \
     -p 127.0.0.1:8080:8080 -p 127.0.0.1:8082:8082 \
     mcr.microsoft.com/dts/dts-emulator:latest
   

   Aspetta che la dashboard in http://localhost:8082 diventi disponibile. Se un altro emulatore già usa queste porte, usa quell'emulatore o fermalo prima di avviarne un secondo.

2. Configura la connessione nel terminale dove eseguirai l'app Go.

   export DTS_CONNECTION_STRING="Endpoint=http://localhost:8080;TaskHub=default;Authentication=None"
   

3. Crea un modulo Go e installa l'SDK.

   mkdir durable-task-go-quickstart
   cd durable-task-go-quickstart
   go mod init example.com/durable-task-go-quickstart
   go get github.com/microsoft/durabletask-go@v1.0.0-beta3
   

Eseguire l'avvio rapido

1. Dalla directory FanOutFanIn passare alla directory Worker per compilare ed eseguire il worker.

   cd Worker
   dotnet build
   dotnet run
   

2. In un terminale separato, dalla FanOutFanIn directory passare alla Client directory per compilare ed eseguire il client.

   cd Client
   dotnet build
   dotnet run
   

Informazioni sull'output

Quando si esegue questo esempio, si riceve l'output sia dai processi di lavoro che da quelli client. Decomprimere ciò che è successo nel codice quando è stato eseguito il progetto.

Produzione del lavoratore

L'output del ruolo di lavoro mostra:

• Registrazione dell'orchestratore e delle attività
• Registrazioni di log ogni volta che viene chiamata un'attività.
• Elaborazione parallela di più elementi di lavoro
• Aggregazione finale dei risultati

Output del client

L'output del client mostra:

• Orchestrazione che inizia con un elenco di elementi di lavoro
• ID univoco dell'istanza di orchestrazione
• Risultati aggregati finali, che mostrano ogni elemento di lavoro e il risultato corrispondente
• Numero totale di elementi elaborati

Output di esempio

Starting Fan-Out Fan-In Pattern - Parallel Processing Client
Using local emulator with no authentication
Starting parallel processing orchestration with 5 work items
Work items: ["Task1","Task2","Task3","LongerTask4","VeryLongTask5"]
Started orchestration with ID: 7f8e9a6b-1c2d-3e4f-5a6b-7c8d9e0f1a2b
Waiting for orchestration to complete...
Orchestration completed with status: Completed
Processing results:
Work item: Task1, Result: 5
Work item: Task2, Result: 5
Work item: Task3, Result: 5
Work item: LongerTask4, Result: 11
Work item: VeryLongTask5, Result: 13
Total items processed: 5

1. Attivare un ambiente virtuale Python.

   • Windows
   • Linux

   python -m venv venv
   /venv/Scripts/activate
   

2. Installare i pacchetti necessari.

   pip install -r requirements.txt
   

3. Avviare il lavoratore.

   python worker.py
   

   Output previsto

   È possibile visualizzare l'output relativo all’avvio del ruolo e all’attesa degli elementi di lavoro.

   Starting Fan Out/Fan In pattern worker...
   Using taskhub: default
   Using endpoint: http://localhost:8080
   Starting gRPC worker that connects to http://localhost:8080
   Successfully connected to http://localhost:8080. Waiting for work items...
   

4. In un nuovo terminale attivare l'ambiente virtuale ed eseguire il client.

   • Windows
   • Linux

   venv/Scripts/activate
   python client.py
   

   È possibile specificare il numero di elementi di lavoro come argomento. Se non viene specificato alcun argomento, l'esempio esegue 10 elementi per impostazione predefinita.

   python client.py [number_of_items]
   

Informazioni sull'output

Quando si esegue questo esempio, si riceve l'output sia dai processi di lavoro che da quelli client. Decomprimere ciò che è successo nel codice quando è stato eseguito il progetto.

Produzione del lavoratore

L'output del ruolo di lavoro mostra:

• Registrazione dell'orchestratore e delle attività.
• Messaggi di stato durante l'elaborazione di ogni elemento di lavoro in parallelo, mostrando che vengono eseguiti contemporaneamente.
• Ritardi casuali per ogni elemento di lavoro (compreso tra 0,5 e 2 secondi) per simulare tempi di elaborazione variabili.
• Messaggio finale che mostra l'aggregazione dei risultati.

Output del client

L'output del client mostra:

• Orchestrazione che inizia con il numero specificato di elementi di lavoro.
• ID dell'istanza di orchestrazione univoca.
• Risultato aggregato finale, che include:
  • Numero totale di elementi elaborati
  • Somma di tutti i risultati (ogni risultato dell'elemento è il quadrato del relativo valore)
  • Media di tutti i risultati

Output di esempio

Starting fan out/fan in orchestration with 10 items
Waiting for 10 parallel tasks to complete
Orchestrator yielded with 10 task(s) and 0 event(s) outstanding.
Processing work item: 1
Processing work item: 2
Processing work item: 10
Processing work item: 9
Processing work item: 8
Processing work item: 7
Processing work item: 6
Processing work item: 5
Processing work item: 4
Processing work item: 3
Orchestrator yielded with 9 task(s) and 0 event(s) outstanding.
Orchestrator yielded with 8 task(s) and 0 event(s) outstanding.
Orchestrator yielded with 7 task(s) and 0 event(s) outstanding.
Orchestrator yielded with 6 task(s) and 0 event(s) outstanding.
Orchestrator yielded with 5 task(s) and 0 event(s) outstanding.
Orchestrator yielded with 4 task(s) and 0 event(s) outstanding.
Orchestrator yielded with 3 task(s) and 0 event(s) outstanding.
Orchestrator yielded with 2 task(s) and 0 event(s) outstanding.
Orchestrator yielded with 1 task(s) and 0 event(s) outstanding.
All parallel tasks completed, aggregating results
Orchestrator yielded with 1 task(s) and 0 event(s) outstanding.
Aggregating results from 10 items
Orchestration completed with status: COMPLETED

Dalla directory fan-out-fan-in, costruisci ed esegui l'applicazione usando Gradle.

./gradlew runFanOutFanInPattern

Informazioni sull'output

Quando si esegue questo esempio, viene visualizzato l'output seguente:

• Registrazione dell'orchestratore e delle attività.
• Messaggi di stato durante l'elaborazione di ogni elemento di lavoro in parallelo, mostrando che vengono eseguiti contemporaneamente.
• Ritardi casuali per ogni elemento di lavoro (compreso tra 0,5 e 2 secondi) per simulare tempi di elaborazione variabili.
• Messaggio finale che mostra l'aggregazione dei risultati.

Decomprimere ciò che è successo nel codice quando è stato eseguito il progetto.

Output di esempio

Starting a Gradle Daemon (subsequent builds will be faster)

> Task :runFanOutFanInPattern
Durable Task worker is connecting to sidecar at localhost:8080.
Started new orchestration instance
Orchestration completed: [Name: 'FanOutFanIn_WordCount', ID: '<id-number>', RuntimeStatus: COMPLETED, CreatedAt: 2025-04-25T15:24:47.170Z, LastUpdatedAt: 2025-04-25T15:24:47.287Z, Input: '["Hello, world!","The quick brown fox jumps over t...', Output: '60']
Output: 60

1. Installare i pacchetti npm necessari.

   npm install
   

2. Avviare il lavoratore.

   npm run worker
   

   Output previsto

   È possibile visualizzare l'output relativo all’avvio del ruolo e all’attesa degli elementi di lavoro.

   Using local emulator with no authentication
   Starting worker...
   Worker is ready and listening for tasks.
   

3. In un nuovo terminale, dalla fan-out-fan-in directory, eseguire il client.

   npm run client
   

   È possibile specificare il numero di elementi di lavoro come argomento. Se non viene specificato alcun argomento, l'esempio esegue 10 elementi per impostazione predefinita.

   npm run client -- 15
   

Informazioni sull'output

Quando si esegue questo esempio, si riceve l'output sia dai processi di lavoro che da quelli client. Decomprimere ciò che è successo nel codice quando è stato eseguito il progetto.

Produzione del lavoratore

L'output del ruolo di lavoro mostra:

• Messaggi di stato durante l'elaborazione di ogni elemento di lavoro in parallelo, mostrando che vengono eseguiti contemporaneamente.
• Ritardi casuali per ogni elemento di lavoro (compreso tra 0,5 e 2 secondi) per simulare tempi di elaborazione variabili.
• Messaggio finale che mostra l'aggregazione dei risultati.

Output del client

L'output del client mostra:

• Orchestrazione che inizia con il numero specificato di elementi di lavoro.
• ID dell'istanza di orchestrazione univoca.
• Risultato aggregato finale, che include:
  • Numero totale di elementi elaborati
  • Somma di tutti i risultati (ogni risultato dell'elemento è il quadrato del relativo valore)
  • Media di tutti i risultati

Output di esempio

Using local emulator with no authentication
Starting fan out/fan in orchestration with 10 items
Started orchestration with ID: abc123-def456-ghi789
Waiting for orchestration to complete...
Processing work item 1 (delay 823ms)
Processing work item 2 (delay 1205ms)
Processing work item 3 (delay 512ms)
Processing work item 4 (delay 1890ms)
Processing work item 5 (delay 645ms)
Processing work item 6 (delay 1102ms)
Processing work item 7 (delay 933ms)
Processing work item 8 (delay 1567ms)
Processing work item 9 (delay 701ms)
Processing work item 10 (delay 1344ms)
Aggregating 10 results...
Orchestration completed with status: COMPLETED
Result: {"totalItems":10,"sum":385,"average":38.5,"results":[...]}

Salva il seguente programma completo come main.go in durable-task-go-quickstart:

package main

import (
	"context"
	"errors"
	"fmt"
	"os"
	"time"

	"github.com/microsoft/durabletask-go/api"
	"github.com/microsoft/durabletask-go/client"
	"github.com/microsoft/durabletask-go/durabletaskscheduler"
	"github.com/microsoft/durabletask-go/task"
)

type summary struct {
	TotalItems int     `json:"total_items"`
	Sum        int64   `json:"sum"`
	Average    float64 `json:"average"`
}

func fanOutFanIn(ctx *task.OrchestrationContext) (any, error) {
	var items []int64
	if err := ctx.GetInput(&items); err != nil {
		return nil, err
	}
	if len(items) > 100 {
		return nil, fmt.Errorf("batch contains more than 100 items")
	}

	pending := make([]task.Task, len(items))
	for i, item := range items {
		pending[i] = ctx.CallActivity("ProcessWorkItem", task.WithActivityInput(item))
	}
	if err := ctx.WhenAll(pending...); err != nil {
		return nil, fmt.Errorf("process batch: %w", err)
	}

	result := summary{TotalItems: len(items)}
	for _, work := range pending {
		var squared int64
		if err := work.Await(&squared); err != nil {
			return nil, err
		}
		result.Sum += squared
	}
	if result.TotalItems != 0 {
		result.Average = float64(result.Sum) / float64(result.TotalItems)
	}
	return result, nil
}

func processWorkItem(ctx task.ActivityContext) (any, error) {
	var item int64
	if err := ctx.GetInput(&item); err != nil {
		return nil, err
	}
	if item < -1_000_000 || item > 1_000_000 {
		return nil, fmt.Errorf("item %d exceeds the sample's safe arithmetic range", item)
	}
	return item * item, nil
}

func run(ctx context.Context) (err error) {
	options, err := durabletaskscheduler.NewOptionsFromConnectionString(
		os.Getenv("DTS_CONNECTION_STRING"))
	if err != nil {
		return err
	}

	registry := task.NewTaskRegistry()
	if err := registry.AddOrchestratorN("FanOutFanIn", fanOutFanIn); err != nil {
		return err
	}
	if err := registry.AddActivityN("ProcessWorkItem", processWorkItem); err != nil {
		return err
	}

	logger := api.DefaultLogger()
	durableClient, err := durabletaskscheduler.NewClient(ctx, options, logger)
	if err != nil {
		return err
	}
	defer func() {
		err = errors.Join(err, durableClient.Close())
	}()

	worker, err := durabletaskscheduler.NewWorker(
		options, registry, logger, client.WithAutoWorkItemFilters())
	if err != nil {
		return err
	}
	if err := worker.Start(ctx); err != nil {
		return err
	}
	defer func() {
		shutdownCtx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
		defer cancel()
		err = errors.Join(err, worker.Shutdown(shutdownCtx))
	}()

	id, err := durableClient.ScheduleNewOrchestration(
		ctx, "FanOutFanIn",
		api.WithInput([]int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}))
	if err != nil {
		return err
	}
	metadata, err := durableClient.WaitForOrchestrationCompletion(
		ctx, id, api.WithFetchPayloads(true))
	if err != nil {
		return err
	}
	if metadata.RuntimeStatus != api.RUNTIME_STATUS_COMPLETED {
		return fmt.Errorf("orchestration ended with %s: %+v",
			metadata.RuntimeStatus, metadata.FailureDetails)
	}
	var result summary
	if err := metadata.ReadOutput(&result); err != nil {
		return err
	}
	fmt.Printf("Instance: %s\nStatus: %s\n", id, metadata.RuntimeStatus)
	fmt.Printf("Items: %d\nSum: %d\nAverage: %.1f\n",
		result.TotalItems, result.Sum, result.Average)
	return nil
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	defer cancel()
	if err := run(ctx); err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
}

Esegui il programma:

go mod tidy
go run .

Esegui go mod tidy dopo aver salvato il codice in modo che Go registri le dipendenze transitive dell'SDK e i relativi checksum.

Il worker e il client vengono eseguiti nello stesso processo. Il worker si arresta e il client si chiude quando la dimostrazione termina. L'istanza rimane nell'emulatore così puoi controllare la sua cronologia.

Informazioni sull'output

Il programma mette al quadrato i numeri da 1 a 10 in parallelo e aggrega i risultati. L'ID dell'istanza varia, ma i valori finali sono:

Instance: <generated-instance-id>
Status: ORCHESTRATION_STATUS_COMPLETED
Items: 10
Sum: 385
Average: 38.5

Un errore di connessione, attività, decodifica o spegnimento produce uno stato di uscita non nullo. Un fallimento del flusso di lavoro terminale non viene considerato un completamento riuscito.

Dopo aver eseguito il progetto in locale, è possibile imparare a distribuire di Azure ospitati in App contenitore di Azure.

Visualizzare lo stato e la cronologia dell'orchestrazione

È possibile visualizzare lo stato e la cronologia dell'orchestrazione tramite la dashboard di pianificazione attività durevoli. Per impostazione predefinita, il dashboard viene eseguito sulla porta 8082.

1. Passare a http://localhost:8082 nel Web browser.
2. Fare clic sull'hub attività predefinito. L'istanza di orchestrazione creata è nell'elenco.
3. Fare clic sull'ID istanza di orchestrazione per visualizzare i dettagli dell'esecuzione, tra cui:
   • Esecuzione parallela di più attività
   • Passaggio di aggregazione della ventola
   • Input e output in ogni passaggio
   • Tempo impiegato per ogni passaggio

Informazioni sulla struttura del codice

Il progetto dei lavoratori

Per dimostrare il modello fan-out/fan-in, l'orchestrazione del progetto di lavoro crea attività parallele e ne attende il completamento. L'orchestratore:

1. Accetta un elenco di elementi di lavoro come input.
2. Esegue il fan-out creando un'attività separata per ogni elemento di lavoro tramite ProcessWorkItemActivity.
3. Esegue tutte le attività in parallelo.
4. Attende il completamento di tutte le attività tramite Task.WhenAll.
5. Esegue il fan-in aggregando tutti i singoli risultati tramite AggregateResultsActivity.
6. Restituisce il risultato aggregato finale al client.

Il progetto di lavoro contiene:

• ParallelProcessingOrchestration.cs: definisce le funzioni dell'agente di orchestrazione e dell'attività in un singolo file.
• Program.cs: configura l'host di lavoro con una corretta gestione delle stringhe di connessione.

ParallelProcessingOrchestration.cs

Usando fan-out/fan-in, l'orchestrazione crea attività parallele e ne attende il completamento.

public override async Task<Dictionary<string, int>> RunAsync(TaskOrchestrationContext context, List<string> workItems)
{
    // Step 1: Fan-out by creating a task for each work item in parallel
    List<Task<Dictionary<string, int>>> processingTasks = new List<Task<Dictionary<string, int>>>();

    foreach (string workItem in workItems)
    {
        // Create a task for each work item (fan-out)
        Task<Dictionary<string, int>> task = context.CallActivityAsync<Dictionary<string, int>>(
            nameof(ProcessWorkItemActivity), workItem);
        processingTasks.Add(task);
    }

    // Step 2: Wait for all parallel tasks to complete
    Dictionary<string, int>[] results = await Task.WhenAll(processingTasks);

    // Step 3: Fan-in by aggregating all results
    Dictionary<string, int> aggregatedResults = await context.CallActivityAsync<Dictionary<string, int>>(
        nameof(AggregateResultsActivity), results);

    return aggregatedResults;
}

Ogni attività viene implementata come classe separata decorata con l'attributo [DurableTask] .

[DurableTask]
public class ProcessWorkItemActivity : TaskActivity<string, Dictionary<string, int>>
{
    // Implementation processes a single work item
}

[DurableTask]
public class AggregateResultsActivity : TaskActivity<Dictionary<string, int>[], Dictionary<string, int>>
{
    // Implementation aggregates individual results
}

Program.cs

Il lavoratore usa Microsoft.Extensions.Hosting per una corretta gestione del ciclo di vita.

using Microsoft.Extensions.Hosting;
//..

builder.Services.AddDurableTaskWorker()
    .AddTasks(registry =>
    {
        registry.AddOrchestrator<ParallelProcessingOrchestration>();
        registry.AddActivity<ProcessWorkItemActivity>();
        registry.AddActivity<AggregateResultsActivity>();
    })
    .UseDurableTaskScheduler(connectionString);

Il progetto del cliente

Il progetto client:

• Usa la stessa logica della stringa di connessione del lavoratore.
• Crea un elenco di elementi di lavoro da elaborare in parallelo.
• Pianifica un'istanza di orchestrazione con l'elenco come input.
• Attende il completamento dell'orchestrazione e visualizza i risultati aggregati.
• Usa WaitForInstanceCompletionAsync per eseguire il polling efficiente.

List<string> workItems = new List<string>
{
    "Task1",
    "Task2",
    "Task3",
    "LongerTask4",
    "VeryLongTask5"
};

// Schedule the orchestration with the work items
string instanceId = await client.ScheduleNewOrchestrationInstanceAsync(
    "ParallelProcessingOrchestration", 
    workItems);

// Wait for completion
var instance = await client.WaitForInstanceCompletionAsync(
    instanceId,
    getInputsAndOutputs: true,
    cts.Token);

worker.py

Per dimostrare il modello fan-out/fan-in, l'orchestrazione del progetto di lavoro crea attività parallele e ne attende il completamento. L'orchestratore:

1. Riceve un elenco di elementi di lavoro come input.
2. Esegue il "fan out" creando attività parallele per ogni elemento di lavoro (chiamando process_work_item per ciascuno di essi).
3. Attende il completamento di tutte le attività usando task.when_all.
4. Quindi esegue il "fan in" aggregando i risultati con l'attività aggregate_results.
5. Il risultato aggregato finale viene restituito al client.

Usando fan-out/fan-in, l'orchestrazione crea attività parallele e ne attende il completamento.

# Orchestrator function
def fan_out_fan_in_orchestrator(ctx, work_items: list) -> dict:
    logger.info(f"Starting fan out/fan in orchestration with {len(work_items)} items")

    # Fan out: Create a task for each work item
    parallel_tasks = []
    for item in work_items:
        parallel_tasks.append(ctx.call_activity("process_work_item", input=item))

    # Wait for all tasks to complete
    logger.info(f"Waiting for {len(parallel_tasks)} parallel tasks to complete")
    results = yield task.when_all(parallel_tasks)

    # Fan in: Aggregate all the results
    logger.info("All parallel tasks completed, aggregating results")
    final_result = yield ctx.call_activity("aggregate_results", input=results)

    return final_result

client.py

Il progetto client:

• Usa la stessa logica della stringa di connessione del lavoratore.
• Crea un elenco di elementi di lavoro da elaborare in parallelo.
• Pianifica un'istanza di orchestrazione con l'elenco come input.
• Attende il completamento dell'orchestrazione e visualizza i risultati aggregati.
• Usa wait_for_orchestration_completion per eseguire il polling efficiente.

# Generate work items (default 10 items if not specified)
count = int(sys.argv[1]) if len(sys.argv) > 1 else 10
work_items = list(range(1, count + 1))

logger.info(f"Starting new fan out/fan in orchestration with {count} work items")

# Schedule a new orchestration instance
instance_id = client.schedule_new_orchestration(
    "fan_out_fan_in_orchestrator", 
    input=work_items
)

logger.info(f"Started orchestration with ID = {instance_id}")

# Wait for orchestration to complete
logger.info("Waiting for orchestration to complete...")
result = client.wait_for_orchestration_completion(
    instance_id,
    timeout=60
)

Per illustrare il modello fan-out/fan-in, l'orchestrazione del progetto FanOutFanInPattern crea attività parallele e ne attende il completamento. L'orchestratore:

1. Accetta un elenco di elementi di lavoro come input.
2. Esegue il fan-out creando un'attività separata per ogni elemento di lavoro tramite callActivity.
3. Esegue tutte le attività in parallelo.
4. Attende il completamento di tutte le attività tramite allOf.
5. Esegue il fan-in aggregando tutti i singoli risultati tramite stream().mapToInt().sum().
6. Restituisce il risultato aggregato finale al client.

Il progetto contiene:

• Processo di lavoro DurableTaskSchedulerWorkerExtensions: definisce le funzioni dell’agente di orchestrazione e delle attività.
• DurableTaskSchedulerClientExtension client: configura l'host di lavoro con una corretta gestione delle stringhe di connessione.

Lavoratore

Usando fan-out/fan-in, l'orchestrazione crea attività parallele e ne attende il completamento.

DurableTaskGrpcWorker worker = DurableTaskSchedulerWorkerExtensions.createWorkerBuilder(connectionString)
    .addOrchestration(new TaskOrchestrationFactory() {
        @Override
        public String getName() { return "FanOutFanIn_WordCount"; }

        @Override
        public TaskOrchestration create() {
            return ctx -> {
                List<?> inputs = ctx.getInput(List.class);
                List<Task<Integer>> tasks = inputs.stream()
                        .map(input -> ctx.callActivity("CountWords", input.toString(), Integer.class))
                        .collect(Collectors.toList());
                List<Integer> allWordCountResults = ctx.allOf(tasks).await();
                int totalWordCount = allWordCountResults.stream().mapToInt(Integer::intValue).sum();
                ctx.complete(totalWordCount);
            };
        }
    })
    .addActivity(new TaskActivityFactory() {
        @Override
        public String getName() { return "CountWords"; }

        @Override
        public TaskActivity create() {
            return ctx -> {
                String input = ctx.getInput(String.class);
                StringTokenizer tokenizer = new StringTokenizer(input);
                return tokenizer.countTokens();
            };
        }
    })
    .build();

// Start the worker
worker.start();

Client

Il progetto client:

• Usa la stessa logica della stringa di connessione del lavoratore.
• Crea un elenco di elementi di lavoro da elaborare in parallelo.
• Pianifica un'istanza di orchestrazione con l'elenco come input.
• Attende il completamento dell'orchestrazione e visualizza i risultati aggregati.
• Usa waitForInstanceCompletion per eseguire il polling efficiente.

DurableTaskClient client = DurableTaskSchedulerClientExtensions.createClientBuilder(connectionString).build();

// The input is an arbitrary list of strings.
List<String> listOfStrings = Arrays.asList(
        "Hello, world!",
        "The quick brown fox jumps over the lazy dog.",
        "If a tree falls in the forest and there is no one there to hear it, does it make a sound?",
        "The greatest glory in living lies not in never falling, but in rising every time we fall.",
        "Always remember that you are absolutely unique. Just like everyone else.");

// Schedule an orchestration which will reliably count the number of words in all the given sentences.
String instanceId = client.scheduleNewOrchestrationInstance(
        "FanOutFanIn_WordCount",
        new NewOrchestrationInstanceOptions().setInput(listOfStrings));
logger.info("Started new orchestration instance: {}", instanceId);

// Block until the orchestration completes. Then print the final status, which includes the output.
OrchestrationMetadata completedInstance = client.waitForInstanceCompletion(
        instanceId,
        Duration.ofSeconds(30),
        true);
logger.info("Orchestration completed: {}", completedInstance);
logger.info("Output: {}", completedInstance.readOutputAs(int.class));

Per illustrare il modello fan-out/fan-in, l'orchestrazione di esempio crea attività parallele e attende il completamento di tutte le attività. L'orchestratore:

1. Riceve un elenco di elementi di lavoro come input.
2. Distribuire creando attività parallele per ogni elemento di lavoro (chiamando processWorkItem per ogni elemento).
3. Attende il completamento di tutte le attività tramite whenAll.
4. Unire le informazioni aggregando i risultati con l'attività aggregateResults.
5. Restituisce il risultato aggregato finale al client.

Il progetto contiene:

• worker.mjs: definisce le funzioni dell'orchestratore e delle funzioni di attività.
• client.mjs: pianifica le orchestrazioni e attende il completamento.

worker.mjs

Usando fan-out/fan-in, l'orchestrazione crea attività parallele e ne attende il completamento.

import { whenAll } from "@microsoft/durabletask-js";
import { createAzureManagedWorkerBuilder } from "@microsoft/durabletask-js-azuremanaged";

// Orchestrator function using generator syntax
const fanOutFanInOrchestrator = async function* (ctx, workItems) {
  const items = Array.isArray(workItems) ? workItems : [];

  // Fan out: schedule parallel activity calls
  const tasks = items.map((item) => ctx.callActivity(processWorkItem, item));

  // Wait for all tasks using whenAll
  const processedResults = yield whenAll(tasks);

  // Fan in: aggregate results
  const finalResult = yield ctx.callActivity(aggregateResults, processedResults);

  return finalResult;
};

// Activity that processes a single work item
const processWorkItem = async (_ctx, workItem) => {
  const normalizedItem = Number(workItem);
  const delayMs = 500 + Math.floor(Math.random() * 1500);
  console.log(`Processing work item ${normalizedItem} (delay ${delayMs}ms)`);

  await new Promise((resolve) => setTimeout(resolve, delayMs));

  return {
    item: normalizedItem,
    result: normalizedItem * normalizedItem,
  };
};

// Activity that aggregates results
const aggregateResults = async (_ctx, results) => {
  const sum = results.reduce((acc, curr) => acc + curr.result, 0);

  return {
    totalItems: results.length,
    sum,
    average: results.length ? sum / results.length : 0,
    results,
  };
};

// Build and start worker
const connectionString = `Endpoint=http://localhost:8080;Authentication=None;TaskHub=default`;
const worker = createAzureManagedWorkerBuilder(connectionString)
  .addOrchestrator(fanOutFanInOrchestrator)
  .addActivity(processWorkItem)
  .addActivity(aggregateResults)
  .build();

await worker.start();

client.mjs

Il progetto client:

• Usa la stessa logica della stringa di connessione del lavoratore.
• Crea un elenco di elementi di lavoro da elaborare in parallelo.
• Pianifica un'istanza di orchestrazione con l'elenco come input.
• Attende il completamento dell'orchestrazione e visualizza i risultati aggregati.
• Usa waitForOrchestrationCompletion per eseguire il polling efficiente.

import { OrchestrationStatus } from "@microsoft/durabletask-js";
import { createAzureManagedClient } from "@microsoft/durabletask-js-azuremanaged";

// Create work items array
const count = 10;
const workItems = Array.from({ length: count }, (_unused, index) => index + 1);

const connectionString = `Endpoint=http://localhost:8080;Authentication=None;TaskHub=default`;
const client = createAzureManagedClient(connectionString);

// Schedule orchestration
const instanceId = await client.scheduleNewOrchestration(
  "fanOutFanInOrchestrator",
  workItems
);

console.log(`Started orchestration with ID: ${instanceId}`);

// Wait for completion
const state = await client.waitForOrchestrationCompletion(instanceId, true, 120);

if (state.runtimeStatus === OrchestrationStatus.COMPLETED) {
  console.log(`Result: ${state.serializedOutput}`);
}

await client.stop();

fanOutFanIn programma ogni attività prima di attendere. ctx.WhenAll Aspetta l'intero lotto, inclusi i compiti rimanenti quando uno fallisce. Await poi decodifica ogni risultato dell'attività. Questo approccio garantisce una concorrenza duratura; l'orchestratore non crea goroutine native.

processWorkItem funziona al di fuori dell'orchestratore e può eseguire il normale Go I/O. Questo esempio utilizza invece l'aritmetica e limita l'input per evitare l'overflow. L'orchestratore limita anche la dimensione del lotto.

run registra i nomi dei compiti, crea connessioni client e worker separate, avvia il worker e programma un'istanza. Richiede payload di output, verifica RuntimeStatuse legge l'output tipizzato. Il suo spegnimento differito utilizza un nuovo contesto così la pulizia può comunque essere eseguita dopo il timeout principale.

Per altri modelli di workflow e limitazioni beta, vedi Sviluppare con l'SDK Durable Task for Go.

Quando hai finito, arresta il contenitore dell'emulatore:

docker stop dtsemulator && docker rm dtsemulator

Se hai iniziato go-dts-emulator per questo quickstart, fermati e rimuovilo:

docker stop go-dts-emulator && docker rm go-dts-emulator

Ora che l'esempio è stato eseguito in locale usando il pianificatore di attività durevole, provare a creare un'utilità di pianificazione e una risorsa dell'hub attività e distribuirle in App Contenitore di Azure.