Quickstart: Een app maken met Durable Task SDK's en Durable Task Scheduler

Vereiste voorwaarden

Voordat u begint:

• Zorg ervoor dat u .NET 10 SDK of hoger hebt.
• Installeer Docker voor het uitvoeren van de emulator.
• Kloon de Durable Task Scheduler GitHub repository om het snelstartvoorbeeld te gebruiken.

• Zorg ervoor dat u Python 3.9+ of hoger hebt.
• Installeer Docker voor het uitvoeren van de emulator.
• Kloon de Durable Task Scheduler GitHub repository om het snelstartvoorbeeld te gebruiken.

• Zorg ervoor dat u Java 8 of 11 hebt.
• Installeer Docker voor het uitvoeren van de emulator.
• Kloon de Durable Task Scheduler GitHub repository om het snelstartvoorbeeld te gebruiken.

• Zorg ervoor dat u Node.js 22 of hoger hebt.
• Installeer Docker voor het uitvoeren van de emulator.
• Kloon de Durable Task Scheduler GitHub repository om het snelstartvoorbeeld te gebruiken.

• Installeer Go 1.25 of hoger.
• Installeer Docker of een compatibele containerruntime.

De emulator voor Durable Task Scheduler instellen

De emulator simuleert de Durable Task Scheduler en een taakhub in een Docker-container, waardoor deze ideaal is voor lokale ontwikkeling.

De voorbeeldcode maakt standaard verbinding met de emulator wanneer er geen omgevingsvariabelen zijn ingesteld.

1. Navigeer vanuit de Azure-Samples/Durable-Task-Scheduler hoofdmap naar de voorbeeldmap van de .NET SDK.

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

2. Haal de Docker-image voor de emulator op.

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

3. Voer de emulator uit. Het kan enkele seconden duren voordat de container gereed is.

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

Omdat de voorbeeldcode automatisch gebruikmaakt van de standaardemulatorinstellingen, hoeft u geen omgevingsvariabelen in te stellen. De standaardemulatorinstellingen voor deze quickstart zijn:

• Eindpunt: http://localhost:8080
• Taakcentrum: default

1. Navigeer vanuit de Azure-Samples/Durable-Task-Scheduler hoofdmap naar de Python SDK-voorbeeldmap.

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

2. Haal de Docker-image voor de emulator op.

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

3. Voer de emulator uit. Het kan enkele seconden duren voordat de container gereed is.

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

Omdat de voorbeeldcode automatisch gebruikmaakt van de standaardemulatorinstellingen, hoeft u geen omgevingsvariabelen in te stellen. De standaardemulatorinstellingen voor deze quickstart zijn:

• Eindpunt: http://localhost:8080
• Taakcentrum: default

1. Navigeer vanuit de Azure-Samples/Durable-Task-Scheduler hoofdmap naar de voorbeeldmap van de Java SDK.

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

2. Haal de Docker-image voor de emulator op.

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

3. Voer de emulator uit. Het kan enkele seconden duren voordat de container gereed is.

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

Omdat de voorbeeldcode automatisch gebruikmaakt van de standaardemulatorinstellingen, hoeft u geen omgevingsvariabelen in te stellen. De standaardemulatorinstellingen voor deze quickstart zijn:

• Eindpunt: http://localhost:8080
• Taakcentrum: default

1. Navigeer vanuit de hoofdmap Azure-Samples/Durable-Task-Scheduler naar de javaScript SDK-voorbeeldmap.

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

2. Haal de Docker-image voor de emulator op.

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

3. Voer de emulator uit. Het kan enkele seconden duren voordat de container gereed is.

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

Omdat de voorbeeldcode automatisch gebruikmaakt van de standaardemulatorinstellingen, hoeft u geen omgevingsvariabelen in te stellen. De standaardemulatorinstellingen voor deze quickstart zijn:

• Eindpunt: http://localhost:8080
• Taakcentrum: default

1. Start een speciale emulator. Bind de poorten naar loopback om de niet-geauthenticeerde emulator lokaal op je machine te houden.

   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
   

   Wacht tot het dashboard http://localhost:8082 beschikbaar is. Als een andere emulator deze poorten al gebruikt, gebruik dan die emulator of stop hem voordat je een tweede begint.

2. Configureer de verbinding in de terminal waar je de Go-app draait.

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

3. Maak een Go-module aan en installeer de 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
   

De quickstart uitvoeren

1. Navigeer vanuit de FanOutFanIn map naar de Worker map om de worker te bouwen en draaien.

   cd Worker
   dotnet build
   dotnet run
   

2. Navigeer in een afzonderlijke terminal, vanuit de FanOutFanIn map, naar de Client map om de client te bouwen en uit te voeren.

   cd Client
   dotnet build
   dotnet run
   

Inzicht in de uitvoer

Wanneer u dit voorbeeld uitvoert, ontvangt u uitvoer van zowel de werknemers- als de clientprocessen. Pak uit wat er in de code is gebeurd toen u het project hebt uitgevoerd.

Arbeidsoutput

De uitvoer van de werknemer toont:

• Registratie van de orchestrator en activiteiten
• Logboekvermeldingen wanneer elke activiteit wordt aangeroepen
• Parallelle verwerking van meerdere werkitems
• Uiteindelijke aggregatie van resultaten

Client-uitvoer

In de clientuitvoer ziet u:

• De orkestratie begint met een lijst van werkitems
• De unieke orkestratie-instantie-ID
• De uiteindelijke samengevoegde resultaten, met elk werkitem en het bijbehorende resultaat
• Totaal aantal verwerkte items

Voorbeelduitvoer

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. Activeer een virtuele Python-omgeving.

   • Windows
   • Linux

   python -m venv venv
   /venv/Scripts/activate
   

2. Installeer de vereiste pakketten.

   pip install -r requirements.txt
   

3. Start de werker.

   python worker.py
   

   Verwachte uitvoer

   U ziet een uitvoer die aangeeft dat de werknemer is gestart en wacht op werkaanwijzingen.

   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. Activeer de virtuele omgeving in een nieuwe terminal en voer de client uit.

   • Windows
   • Linux

   venv/Scripts/activate
   python client.py
   

   U kunt het aantal werkitems opgeven als argument. Als er geen argument is opgegeven, worden in het voorbeeld standaard 10 items uitgevoerd.

   python client.py [number_of_items]
   

Inzicht in de uitvoer

Wanneer u dit voorbeeld uitvoert, ontvangt u uitvoer van zowel de werknemers- als de clientprocessen. Pak uit wat er in de code is gebeurd toen u het project hebt uitgevoerd.

Arbeidsoutput

De uitvoer van de werknemer toont:

• Registratie van de orchestrator en activiteiten.
• Statusberichten bij het parallel verwerken van elk werkitem, waarin wordt aangegeven dat ze gelijktijdig worden uitgevoerd.
• Willekeurige vertragingen voor elk werkitem (tussen 0,5 en 2 seconden) om verschillende verwerkingstijden te simuleren.
• Een laatste bericht met de aggregatie van resultaten.

Client-uitvoer

In de clientuitvoer ziet u:

• De orkestratie begint met het opgegeven aantal werkitems.
• De unieke ID van de orkestratie-instantie.
• Het uiteindelijke geaggregeerde resultaat, waaronder:
  • Het totale aantal verwerkte items
  • De som van alle resultaten (elk itemresultaat is het kwadraat van de waarde)
  • Het gemiddelde van alle resultaten

Voorbeelduitvoer

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

Bouw en voer de toepassing uit vanuit de fan-out-fan-in map met gradle.

./gradlew runFanOutFanInPattern

Inzicht in de uitvoer

Wanneer u dit voorbeeld uitvoert, ontvangt u uitvoer met:

• Registratie van de orchestrator en activiteiten.
• Statusberichten bij het parallel verwerken van elk werkitem, waarin wordt aangegeven dat ze gelijktijdig worden uitgevoerd.
• Willekeurige vertragingen voor elk werkitem (tussen 0,5 en 2 seconden) om verschillende verwerkingstijden te simuleren.
• Een laatste bericht met de aggregatie van resultaten.

Pak uit wat er in de code is gebeurd toen u het project hebt uitgevoerd.

Voorbeelduitvoer

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. Installeer de vereiste npm-pakketten.

   npm install
   

2. Start de werker.

   npm run worker
   

   Verwachte uitvoer

   U ziet een uitvoer die aangeeft dat de werknemer is gestart en wacht op werkaanwijzingen.

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

3. Voer in een nieuwe terminal vanuit de fan-out-fan-in map de client uit.

   npm run client
   

   U kunt het aantal werkitems opgeven als argument. Als er geen argument is opgegeven, worden in het voorbeeld standaard 10 items uitgevoerd.

   npm run client -- 15
   

Inzicht in de uitvoer

Wanneer u dit voorbeeld uitvoert, ontvangt u uitvoer van zowel de werknemers- als de clientprocessen. Pak uit wat er in de code is gebeurd toen u het project hebt uitgevoerd.

Arbeidsoutput

De uitvoer van de werknemer toont:

• Statusberichten bij het parallel verwerken van elk werkitem, waarin wordt aangegeven dat ze gelijktijdig worden uitgevoerd.
• Willekeurige vertragingen voor elk werkitem (tussen 0,5 en 2 seconden) om verschillende verwerkingstijden te simuleren.
• Een laatste bericht met de aggregatie van resultaten.

Client-uitvoer

In de clientuitvoer ziet u:

• De orkestratie begint met het opgegeven aantal werkitems.
• De unieke ID van de orkestratie-instantie.
• Het uiteindelijke geaggregeerde resultaat, waaronder:
  • Het totale aantal verwerkte items
  • De som van alle resultaten (elk itemresultaat is het kwadraat van de waarde)
  • Het gemiddelde van alle resultaten

Voorbeelduitvoer

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":[...]}

Sla het volgende volledige programma op als 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)
	}
}

Start het programma:

go mod tidy
go run .

Voer go mod tidy uit nadat de code is opgeslagen, zodat Go de transitieve afhankelijkheden en checksums van de SDK registreert.

De worker en de cliënt worden in hetzelfde proces uitgevoerd. De worker wordt leeggemaakt en de client sluit zodra de demonstratie is afgelopen. De instantie blijft in de emulator zodat je de geschiedenis kunt bekijken.

Inzicht in de uitvoer

Het programma kwadrateert de cijfers 1 tot en met 10 parallel en voegt de resultaten samen. De instance ID varieert, maar de uiteindelijke waarden zijn:

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

Een verbindings-, activiteit-, decoder- of afsluitfout veroorzaakt een niet-nul exitstatus. Een terminale workflowfout wordt niet als succesvolle afronding beschouwd.

Nu u het project lokaal hebt uitgevoerd, leert u hoe u een implementatie uitvoert naar Azure, gehost in Azure Container Apps.

Orchestratiestatus en -geschiedenis weergeven

U kunt de orchestratiestatus en -geschiedenis bekijken via het Durable Task Scheduler-dashboard. Standaard wordt het dashboard uitgevoerd op poort 8082.

1. Navigeer naar http://localhost:8082 in uw webbrowser.
2. Klik op de standaardtaakhub. Het orchestratie-exemplaar dat u hebt gemaakt, bevindt zich in de lijst.
3. Klik op de orkestratie-instantie-ID om de uitvoeringsdetails weer te geven, zoals:
   • De parallelle uitvoering van meerdere activiteitstaken
   • De aggregatiestap voor fan-in
   • De invoer en de uitvoer op elk niveau
   • De tijd die nodig is voor elke stap

Inzicht in de codestructuur

Het werknemersproject

Om het fan-out-/fan-in-patroon te demonstreren, maakt de werkrolprojectindeling parallelle activiteitstaken en wacht totdat alles is voltooid. De orkestrator

1. Hiermee wordt een lijst met werkitems als invoer gebruikt.
2. Waaiert uit door een afzonderlijke taak te maken voor elk werkitem met behulp van ProcessWorkItemActivity.
3. Hiermee worden alle taken parallel uitgevoerd.
4. Wacht tot alle taken zijn voltooid met behulp van Task.WhenAll.
5. Fans in door alle afzonderlijke resultaten te aggregeren met behulp van AggregateResultsActivity.
6. Retourneert het uiteindelijke geaggregeerde resultaat naar de client.

Het werknemersproject bevat:

• ParallelProcessingOrchestration.cs: Definieert de orchestrator- en activiteitsfuncties in één bestand.
• Program.cs: configureert de worker host met de juiste verwerking van de verbindingsreeks.

ParallelProcessingOrchestration.cs

Met behulp van fan-out/fan-in creëert de orkestratie parallelle activiteitstaken en wacht tot alle taken zijn voltooid.

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

Elke activiteit wordt geïmplementeerd als een afzonderlijke klasse die is ingericht met het [DurableTask] kenmerk.

[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

De werknemer gebruikt Microsoft.Extensions.Hosting voor het juiste levenscyclusbeheer.

using Microsoft.Extensions.Hosting;
//..

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

Het clientproject

Het clientproject:

• Gebruikt dezelfde verbindingsreeks-logica als de worker.
• Hiermee maakt u een lijst met werkitems die parallel moeten worden verwerkt.
• Hiermee plant u een orchestratie-instantiatie met de lijst als invoer.
• Wacht totdat de orchestratie is voltooid en geeft de geaggregeerde resultaten weer.
• Gebruikt WaitForInstanceCompletionAsync voor efficiënte polling.

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

Om het fan-out-/fan-in-patroon te demonstreren, maakt de werkrolprojectindeling parallelle activiteitstaken en wacht totdat alles is voltooid. De orkestrator

1. Hiermee ontvangt u een lijst met werkitems als invoer.
2. Het verspreidt zich door parallelle taken te maken voor elk werkitem (roept process_work_item aan voor elk werkitem).
3. Er wordt gewacht tot alle taken zijn voltooid met behulp van task.when_all.
4. Vervolgens wordt er samengevoegd door de resultaten met de aggregate_results activiteit te combineren.
5. Het uiteindelijke geaggregeerde resultaat wordt geretourneerd aan de client.

Met behulp van fan-out/fan-in creëert de orkestratie parallelle activiteitstaken en wacht tot alle taken zijn voltooid.

# 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

Het clientproject:

• Gebruikt dezelfde verbindingsreeks-logica als de worker.
• Hiermee maakt u een lijst met werkitems die parallel moeten worden verwerkt.
• Hiermee plant u een orchestratie-instantiatie met de lijst als invoer.
• Wacht totdat de orchestratie is voltooid en geeft de geaggregeerde resultaten weer.
• Gebruikt wait_for_orchestration_completion voor efficiënte polling.

# 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
)

Om het fan-out-/fan-in-patroon te demonstreren, creëert de FanOutFanInPattern projectorkestratie parallelle activiteitstaken en wacht tot alles is voltooid. De orkestrator

1. Hiermee wordt een lijst met werkitems als invoer gebruikt.
2. Waaiert uit door een afzonderlijke taak te maken voor elk werkitem met behulp van callActivity.
3. Hiermee worden alle taken parallel uitgevoerd.
4. Wacht tot alle taken zijn voltooid met behulp van allOf.
5. Fans in door alle afzonderlijke resultaten te aggregeren met behulp van stream().mapToInt().sum().
6. Retourneert het uiteindelijke geaggregeerde resultaat naar de client.

Het project bevat:

• DurableTaskSchedulerWorkerExtensions worker: Definieert de orchestrator- en activiteitsfuncties.
• DurableTaskSchedulerClientExtension client: hiermee stelt u de worker host in met de juiste afhandeling van de verbindingsreeks.

Werknemer

Met behulp van fan-out/fan-in creëert de orkestratie parallelle activiteitstaken en wacht tot alle taken zijn voltooid.

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

Het clientproject:

• Gebruikt dezelfde verbindingsreeks-logica als de worker.
• Hiermee maakt u een lijst met werkitems die parallel moeten worden verwerkt.
• Hiermee plant u een orchestratie-instantiatie met de lijst als invoer.
• Wacht totdat de orchestratie is voltooid en geeft de geaggregeerde resultaten weer.
• Gebruikt waitForInstanceCompletion voor efficiënte polling.

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));

Om het fan-out-/fan-in-patroon te demonstreren, creëert de voorbeeldorkestratie parallelle activiteitstaken en wacht totdat alle taken zijn voltooid. De orkestrator

1. Hiermee ontvangt u een lijst met werkitems als invoer.
2. Verspreidt zich door parallelle taken te maken voor elk werkitem (waarbij processWorkItem voor elk wordt aangeroepen).
3. Wacht tot alle taken zijn voltooid met behulp van whenAll.
4. Fans in door de resultaten te aggregeren met de aggregateResults activiteit.
5. Retourneert het uiteindelijke geaggregeerde resultaat naar de client.

Het project bevat:

• worker.mjs: Definieert de orchestrator- en activiteitsfuncties.
• client.mjs: Plan orkestraties en wacht op voltooiing.

worker.mjs

Met behulp van fan-out/fan-in creëert de orkestratie parallelle activiteitstaken en wacht tot alle taken zijn voltooid.

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

Het clientproject:

• Gebruikt dezelfde verbindingsreeks-logica als de worker.
• Hiermee maakt u een lijst met werkitems die parallel moeten worden verwerkt.
• Hiermee plant u een orchestratie-instantiatie met de lijst als invoer.
• Wacht totdat de orchestratie is voltooid en geeft de geaggregeerde resultaten weer.
• Gebruikt waitForOrchestrationCompletion voor efficiënte polling.

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 plant elke activiteit voordat je wacht. ctx.WhenAll wacht op de hele batch, inclusief de resterende taken als er één faalt. Await decodeert vervolgens elk activiteitsresultaat. Deze aanpak biedt duurzame concurrency; de orchestrator maakt geen native goroutines aan.

processWorkItem draait buiten de orchestrator en kan gewone Go I/O uitvoeren. Dit voorbeeld gebruikt in plaats daarvan rekenkunde en begrenst de invoer om overflow te voorkomen. De orkestrator begrenst ook de batchgrootte.

run registreert de taaknamen, maakt aparte client- en worker-verbindingen, start de worker en plant één exemplaar. Het vraagt outputpayloads op, controleert RuntimeStatusen leest getypte output. De uitgestelde afsluiting gebruikt een nieuwe context, zodat de opruiming nog steeds kan doorgaan na de hoofd-timeout.

Voor andere workflowpatronen en bètabeperkingen, zie Develop with the Durable Task SDK for Go.

Stop de emulatorcontainer wanneer u klaar bent:

docker stop dtsemulator && docker rm dtsemulator

Als je bent begonnen go-dts-emulator met deze quickstart, stop dan en verwijder hem:

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

Nu u het voorbeeld lokaal hebt uitgevoerd met behulp van de Durable Task Scheduler-emulator, kunt u een scheduler- en task hub-resource maken en implementeren in Azure Container Apps.