Snabbstart: Skapa en app med Durable Task SDK:er och Durable Task Scheduler

Förutsättningar

Innan du börjar:

• Kontrollera att du har .NET 10 SDK eller senare.
• Installera Docker för att köra emulatorn.
• Klona Durable Task Scheduler GitHub-lagringsplatsen för att använda snabbstartsexemplet.

• Kontrollera att du har Python 3.9+ eller senare.
• Installera Docker för att köra emulatorn.
• Klona Durable Task Scheduler GitHub-lagringsplatsen för att använda snabbstartsexemplet.

• Kontrollera att du har Java 8 eller 11.
• Installera Docker för att köra emulatorn.
• Klona Durable Task Scheduler GitHub-lagringsplatsen för att använda snabbstartsexemplet.

• Kontrollera att du har Node.js 22 eller senare.
• Installera Docker för att köra emulatorn.
• Klona Durable Task Scheduler GitHub-lagringsplatsen för att använda snabbstartsexemplet.

• Installera Go 1.25 eller senare.
• Installera Docker eller en kompatibel container-runtime.

Konfigurera Durable Task Scheduler-emulatorn

Emulatorn simulerar Durable Task Scheduler och en aktivitetshubb i en Docker-container, vilket gör den idealisk för lokal utveckling.

Exempelkoden ansluter till emulatorn som standard när inga miljövariabler har angetts.

1. Från rotkatalogen Azure-Samples/Durable-Task-Scheduler går du till exempelkatalogen .NET SDK.

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

2. Hämta Docker-avbildningen för emulatorn.

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

3. Kör emulatorn. Det kan ta några sekunder innan containern är klar.

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

Eftersom exempelkoden automatiskt använder standardinställningarna för emulatorn behöver du inte ange några miljövariabler. Standardinställningarna för emulatorn för den här snabbstarten är:

• Slutpunkt: http://localhost:8080
• Aktivitetshubben: default

1. Från rotkatalogen Azure-Samples/Durable-Task-Scheduler går du till exempelkatalogen Python SDK.

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

2. Hämta Docker-avbildningen för emulatorn.

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

3. Kör emulatorn. Det kan ta några sekunder innan containern är klar.

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

Eftersom exempelkoden automatiskt använder standardinställningarna för emulatorn behöver du inte ange några miljövariabler. Standardinställningarna för emulatorn för den här snabbstarten är:

• Slutpunkt: http://localhost:8080
• Aktivitetshubben: default

1. Från rotkatalogen Azure-Samples/Durable-Task-Scheduler går du till exempelkatalogen Java SDK.

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

2. Hämta Docker-avbildningen för emulatorn.

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

3. Kör emulatorn. Det kan ta några sekunder innan containern är klar.

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

Eftersom exempelkoden automatiskt använder standardinställningarna för emulatorn behöver du inte ange några miljövariabler. Standardinställningarna för emulatorn för den här snabbstarten är:

• Slutpunkt: http://localhost:8080
• Aktivitetshubben: default

1. Från rotkatalogen Azure-Samples/Durable-Task-Scheduler navigerar du till JavaScript SDK-exempelkatalogen.

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

2. Hämta Docker-avbildningen för emulatorn.

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

3. Kör emulatorn. Det kan ta några sekunder innan containern är klar.

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

Eftersom exempelkoden automatiskt använder standardinställningarna för emulatorn behöver du inte ange några miljövariabler. Standardinställningarna för emulatorn för den här snabbstarten är:

• Slutpunkt: http://localhost:8080
• Aktivitetshubben: default

1. Starta en dedikerad emulator. Bind portarna till loopback för att hålla den oautentiserade emulatorn lokal på din dator.

   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
   

   Vänta på att instrumentpanel http://localhost:8082 ska bli tillgänglig. Om en annan emulator redan använder dessa portar, använd den emulatorn eller stoppa den innan du startar en andra emulator.

2. Konfigurera anslutningen i terminalen där du ska köra Go-appen.

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

3. Skapa en Go-modul och installera SDK:n.

   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
   

Kör snabbstarten

1. Från katalogen FanOutFanIn går du till Worker katalogen för att skapa och köra arbetaren.

   cd Worker
   dotnet build
   dotnet run
   

2. I en separat terminal, från FanOutFanIn katalogen, navigerar du till Client katalogen för att skapa och köra klienten.

   cd Client
   dotnet build
   dotnet run
   

Förstå utdata

När du kör det här exemplet får du utdata från både arbets- och klientprocesserna. Packa upp det som hände i koden när du körde projektet.

Arbetsproduktivitet

Arbetsutdata visar:

• Registrering av orkestreraren och aktiviteter
• Loggposter när varje aktivitet anropas
• Parallell bearbetning av flera arbetsobjekt
• Slutlig sammanställning av resultat

Klientens resultat

Klientutdata visar:

• Orkestreringen börjar med en lista över arbetsobjekt
• Det unika orkestreringsinstans-ID:t
• De slutliga aggregerade resultaten, som visar varje arbetsobjekt och dess motsvarande resultat
• Totalt antal bearbetade objekt

Exempel på utdata

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. Aktivera en virtuell Python-miljö.

   • Windows
   • Linux

   python -m venv venv
   /venv/Scripts/activate
   

2. Installera de paket som krävs.

   pip install -r requirements.txt
   

3. Starta arbetaren.

   python worker.py
   

   Förväntade utdata

   Du kan se utdata som visar att arbetaren har startat och väntar på arbetsuppgifter.

   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. I en ny terminal aktiverar du den virtuella miljön och kör klienten.

   • Windows
   • Linux

   venv/Scripts/activate
   python client.py
   

   Du kan ange antalet arbetsobjekt som argument. Om inget argument anges körs 10 objekt som standard i exemplet.

   python client.py [number_of_items]
   

Förstå utdata

När du kör det här exemplet får du utdata från både arbets- och klientprocesserna. Packa upp det som hände i koden när du körde projektet.

Arbetsproduktivitet

Arbetsutdata visar:

• Registrering av orkestreraren och aktiviteter.
• Statusmeddelanden vid bearbetning av varje arbetsobjekt parallellt, som visar att de körs samtidigt.
• Slumpmässiga fördröjningar för varje arbetsobjekt (mellan 0,5 och 2 sekunder) för att simulera varierande bearbetningstider.
• Ett slutligt meddelande som visar sammanställningen av resultat.

Klientens resultat

Klientutdata visar:

• Orkestreringen börjar med det angivna antalet arbetsobjekt.
• Det unika orkestreringsinstans-ID:t.
• Det slutliga aggregerade resultatet, som innehåller:
  • Det totala antalet bearbetade objekt
  • Summan av alla resultat (varje objektresultat är kvadraten för dess värde)
  • Medelvärdet av alla resultat

Exempel på utdata

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

Från katalogen fan-out-fan-in skapar och kör du programmet med Hjälp av Gradle.

./gradlew runFanOutFanInPattern

Förstå utdata

När du kör det här exemplet får du utdata som visar:

• Registrering av orkestreraren och aktiviteter.
• Statusmeddelanden vid bearbetning av varje arbetsobjekt parallellt, som visar att de körs samtidigt.
• Slumpmässiga fördröjningar för varje arbetsobjekt (mellan 0,5 och 2 sekunder) för att simulera varierande bearbetningstider.
• Ett slutligt meddelande som visar sammanställningen av resultat.

Packa upp det som hände i koden när du körde projektet.

Exempel på utdata

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. Installera nödvändiga npm-paket.

   npm install
   

2. Starta arbetaren.

   npm run worker
   

   Förväntade utdata

   Du kan se utdata som visar att arbetaren har startat och väntar på arbetsuppgifter.

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

3. Kör klienten från fan-out-fan-in katalogen i en ny terminal.

   npm run client
   

   Du kan ange antalet arbetsobjekt som argument. Om inget argument anges körs 10 objekt som standard i exemplet.

   npm run client -- 15
   

Förstå utdata

När du kör det här exemplet får du utdata från både arbets- och klientprocesserna. Packa upp det som hände i koden när du körde projektet.

Arbetsproduktivitet

Arbetsutdata visar:

• Statusmeddelanden vid bearbetning av varje arbetsobjekt parallellt, som visar att de körs samtidigt.
• Slumpmässiga fördröjningar för varje arbetsobjekt (mellan 0,5 och 2 sekunder) för att simulera varierande bearbetningstider.
• Ett slutligt meddelande som visar sammanställningen av resultat.

Klientens resultat

Klientutdata visar:

• Orkestreringen börjar med det angivna antalet arbetsobjekt.
• Det unika orkestreringsinstans-ID:t.
• Det slutliga aggregerade resultatet, som innehåller:
  • Det totala antalet bearbetade objekt
  • Summan av alla resultat (varje objektresultat är kvadraten för dess värde)
  • Medelvärdet av alla resultat

Exempel på utdata

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

Spara följande kompletta program som main.go i 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)
	}
}

Kör programmet:

go mod tidy
go run .

Kör go mod tidy efter att koden sparats så att Go registrerar SDK:ns transitiva beroenden och kontrollsummor.

Arbetaren och klienten körs i samma process. Worker-processen tömmer och klienten stänger när demonstrationen är slut. Instansen finns kvar i emulatorn så att du kan granska dess historik.

Förstå utdata

Programmet kvadrerar siffrorna 1 till 10 parallellt och aggregerar resultaten. instans-ID:t varierar, men de slutliga värdena är:

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

Ett anslutnings-, aktivitets-, avkodnings- eller avstängningsfel ger en icke-noll avslutningsstatus. Ett terminalt arbetsflödesfel behandlas inte som ett lyckat slutförande.

Nu när du har kört projektet lokalt lär du dig att distribuera till Azure i Azure Container Apps.

Visa orkestreringsstatus och historik

Du kan visa orkestreringsstatus och historik via instrumentpanelen Durable Task Scheduler. Som standard körs instrumentpanelen på port 8082.

1. Navigera till http://localhost:8082 i webbläsaren.
2. Klicka på standard aktivitetshubben. Orkestreringsinstansen som du skapade finns i listan.
3. Klicka på orkestreringsinstans-ID:t för att visa körningsinformationen, som omfattar:
   • Parallell körning av flera aktivitetsaktiviteter
   • Aggregeringssteget för insamlare
   • Indata och utdata i varje steg
   • Den tid det tar för varje steg

Förstå kodstrukturen

Arbetsprojektet

För att demonstrera fan-out/fan-in-mönstret skapar orkestreringen av arbetsprocessen parallella aktiviteter och väntar på att alla ska slutföras. Dirigeraren:

1. Tar en lista över arbetsobjekt som indata.
2. Sprider ut sig genom att skapa en separat uppgift för varje arbetsuppgift med hjälp av ProcessWorkItemActivity.
3. Kör alla uppgifter parallellt.
4. Väntar på att alla uppgifter ska slutföras med hjälp av Task.WhenAll.
5. Samlar in genom att aggregera alla enskilda resultat med hjälp av AggregateResultsActivity.
6. Returnerar det slutliga aggregerade resultatet till klienten.

Arbetsprojektet innehåller:

• ParallelProcessingOrchestration.cs: Definierar orkestrerings- och aktivitetsfunktionerna i en enda fil.
• Program.cs: Konfigurerar arbetsnoden med korrekt hantering av anslutningssträngar.

ParallelProcessingOrchestration.cs

Med fan-out/fan-in skapar orkestreringen parallella uppgifter och väntar på att alla ska slutföras.

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

Varje aktivitet implementeras som en separat klass som är dekorerad med [DurableTask] attributet .

[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

Arbetaren använder Microsoft.Extensions.Hosting för korrekt livscykelhantering.

using Microsoft.Extensions.Hosting;
//..

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

Klientprojektet

Klientprojektet:

• Använder samma anslutningssträngslogik som arbetaren.
• Skapar en lista över arbetsobjekt som ska bearbetas parallellt.
• Schemalägger en orkestreringsinstans med listan som indata.
• Väntar på att orkestreringen ska slutföras och visar de aggregerade resultaten.
• "WaitForInstanceCompletionAsync används för effektiv avsökning."

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

För att demonstrera fan-out/fan-in-mönstret skapar orkestreringen av arbetsprocessen parallella aktiviteter och väntar på att alla ska slutföras. Dirigeraren:

1. Tar emot en lista över arbetsobjekt som indata.
2. Den "fläktar ut" genom att skapa parallella uppgifter för varje arbetsobjekt (anropa process_work_item för var och en).
3. Den väntar på att alla uppgifter ska slutföras med hjälp av task.when_all.
4. Den samlar sedan in genom att aggregera resultaten med aggregate_results-aktiviteten.
5. Det slutliga aggregerade resultatet returneras till klienten.

Med fan-out/fan-in skapar orkestreringen parallella uppgifter och väntar på att alla ska slutföras.

# 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

Klientprojektet:

• Använder samma anslutningssträngslogik som arbetaren.
• Skapar en lista över arbetsobjekt som ska bearbetas parallellt.
• Schemalägger en orkestreringsinstans med listan som indata.
• Väntar på att orkestreringen ska slutföras och visar de aggregerade resultaten.
• "wait_for_orchestration_completion används för effektiv avsökning."

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

För att demonstrera fan-out/fan-in-mönstret skapar projektorkestreringen parallella aktivitetsuppgifter och väntar på att alla ska slutföras. Dirigeraren:

1. Tar en lista över arbetsobjekt som indata.
2. Sprider ut sig genom att skapa en separat uppgift för varje arbetsuppgift med hjälp av callActivity.
3. Kör alla uppgifter parallellt.
4. Väntar på att alla uppgifter ska slutföras med hjälp av allOf.
5. Samlar in genom att aggregera alla enskilda resultat med hjälp av stream().mapToInt().sum().
6. Returnerar det slutliga aggregerade resultatet till klienten.

Projektet innehåller:

• DurableTaskSchedulerWorkerExtensions worker: Definierar orkestrerings- och aktivitetsfunktionerna.
• DurableTaskSchedulerClientExtension klient: Konfigurerar arbetarvärd med korrekt hantering av anslutningssträngar.

Arbetare

Med fan-out/fan-in skapar orkestreringen parallella uppgifter och väntar på att alla ska slutföras.

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

Klientprojektet:

• Använder samma anslutningssträngslogik som arbetaren.
• Skapar en lista över arbetsobjekt som ska bearbetas parallellt.
• Schemalägger en orkestreringsinstans med listan som indata.
• Väntar på att orkestreringen ska slutföras och visar de aggregerade resultaten.
• "waitForInstanceCompletion används för effektiv avsökning."

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

För att demonstrera mönstret fan-out/fan-in skapar exempelorkestreringen parallella aktiviteter och väntar tills alla slutförs. Dirigeraren:

1. Tar emot en lista över arbetsobjekt som indata.
2. Delar upp genom att skapa parallella uppgifter för varje enskilt arbetsobjekt (anropa processWorkItem för var och en).
3. Väntar på att alla uppgifter ska slutföras med hjälp av whenAll.
4. Fläktar in genom att aggregera resultatet med aggregateResults aktiviteten.
5. Returnerar det slutliga aggregerade resultatet till klienten.

Projektet innehåller:

• worker.mjs: Definierar orkestrerings- och aktivitetsfunktionerna.
• client.mjs: Schemalägger orkestreringar och väntar på slutförande.

worker.mjs

Med fan-out/fan-in skapar orkestreringen parallella uppgifter och väntar på att alla ska slutföras.

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

Klientprojektet:

• Använder samma anslutningssträngslogik som arbetaren.
• Skapar en lista över arbetsobjekt som ska bearbetas parallellt.
• Schemalägger en orkestreringsinstans med listan som indata.
• Väntar på att orkestreringen ska slutföras och visar de aggregerade resultaten.
• "waitForOrchestrationCompletion används för effektiv avsökning."

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 Schemalägger varje aktivitet innan den väntar. ctx.WhenAll väntar på hela omgången, inklusive återstående uppgifter när en misslyckas. Await avkodar sedan varje aktivitetsresultat. Detta tillvägagångssätt ger beständig samtidighet; orkestreraren skapar inte inbyggda goroutines.

processWorkItem kör utanför orkestratören och kan utföra vanliga Go I/O. Detta exempel använder istället aritmetik och begränsar indata för att undvika överflöde. Orkestratören avgränsar också batchstorleken.

run registrerar uppgiftsnamnen, skapar separata klient- och worker-anslutningar, startar arbetaren och schemalägger en instans. Den begär utdatanyttolaster, kontrollerar RuntimeStatus och läser av typad utdata. Dess uppskjutna avstängning använder en ny kontext så att rensningen fortfarande kan köras efter huvudtimeouten.

För andra arbetsflödesmönster och betabegränsningar, se Utveckla med Durable Task SDK för Go.

Stoppa emulatorcontainern när du är klar:

docker stop dtsemulator && docker rm dtsemulator

Om du startade go-dts-emulator för den här snabbstarten, stoppa och ta bort den:

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

Nu när du har kört exemplet lokalt med hjälp av Durable Task Scheduler-emulatorn kan du prova att skapa en scheduler- och aktivitetshubbresurs och distribuera till Azure Container Apps.