Förgrenings-/sammanfogningsmönster

• Slutför snabbstartsartikeln
• Klona eller ladda ner sample-projektet från GitHub
  • Isolerade modellprover
  • Exempel på modeller under processen

• Slutför snabbstartsartikeln
• Clone eller ladda ned exempelprojektet från GitHub

• Slutför snabbstartsartikeln
• Clone eller ladda ned exempelprojektet från GitHub

• Slutför snabbstartsartikeln

Följande kod visar hur du implementerar orchestrator-funktionen:

using System.IO;
using System.Linq;
using System.Threading.Tasks;
using Microsoft.Azure.Functions.Worker;
using Microsoft.DurableTask;

namespace SampleApp;

public static class BackupSiteContent
{
    [Function("E2_BackupSiteContent")]
    public static async Task<long> Run(
        [OrchestrationTrigger] TaskOrchestrationContext context)
    {
        string rootDirectory = context.GetInput<string>()?.Trim();
        if (string.IsNullOrEmpty(rootDirectory))
        {
            rootDirectory = Directory.GetParent(typeof(BackupSiteContent).Assembly.Location)!.FullName;
        }

        string[] files = await context.CallActivityAsync<string[]>("E2_GetFileList", rootDirectory);

        Task<long>[] tasks = files
            .Select(file => context.CallActivityAsync<long>("E2_CopyFileToBlob", file))
            .ToArray();

        long[] results = await Task.WhenAll(tasks);
        return results.Sum();
    }
}

[FunctionName("E2_BackupSiteContent")]
public static async Task<long> Run(
    [OrchestrationTrigger] IDurableOrchestrationContext backupContext)
{
    string rootDirectory = backupContext.GetInput<string>()?.Trim();
    if (string.IsNullOrEmpty(rootDirectory))
    {
        rootDirectory = Directory.GetParent(typeof(BackupSiteContent).Assembly.Location).FullName;
    }

    string[] files = await backupContext.CallActivityAsync<string[]>(
        "E2_GetFileList",
        rootDirectory);

    var tasks = new Task<long>[files.Length];
    for (int i = 0; i < files.Length; i++)
    {
        tasks[i] = backupContext.CallActivityAsync<long>(
            "E2_CopyFileToBlob",
            files[i]);
    }

    await Task.WhenAll(tasks);

    long totalBytes = tasks.Sum(t => t.Result);
    return totalBytes;
}

Funktionen använder standard-function.json för orchestrator-funktioner.

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "context",
      "type": "orchestrationTrigger",
      "direction": "in"
    }
  ]
}

Följande kod visar hur du implementerar orchestrator-funktionen:

import azure.functions as func
import azure.durable_functions as df


def orchestrator_function(context: df.DurableOrchestrationContext):

    root_directory: str = context.get_input()

    if not root_directory:
        raise Exception("A directory path is required as input")

    files = yield context.call_activity("E2_GetFileList", root_directory)
    tasks = []
    for file in files:
        tasks.append(context.call_activity("E2_CopyFileToBlob", file))
    
    results = yield context.task_all(tasks)
    total_bytes = sum(results)
    return total_bytes

main = df.Orchestrator.create(orchestrator_function)

Lägg märke till yield context.task_all(tasks); raden. Koden ger inte de enskilda anropen till E2_CopyFileToBlob, så de körs parallellt. När orkestreraren skickar aktivitetsmatrisen till context.task_allreturneras en aktivitet som inte slutförs förrän alla kopieringsåtgärder har slutförts. Om du är bekant med asyncio.gather i Python är det här konceptet inte nytt för dig. Med Durable Functions-tillägget körs dessa uppgifter på flera virtuella datorer samtidigt, och slut-till-slut-körningen är tålig mot processåtervinning.

När orkestreraren avkastar context.task_all slutförs alla funktionsanrop och returvärden returneras. Varje anrop till E2_CopyFileToBlob returnerar antalet uppladdade byte, så du kan beräkna det totala antalet byte genom att lägga ihop alla returvärdena.

Orchestratorn får en lista med filer och fördelar sedan arbetet för att kopiera varje fil till bloblagringen parallellt med hjälp av -NoWait. När alla parallella uppgifter är klara summeras resultaten.

param($Context)

$rootDirectory = $Context.Input

# Get all files in the directory
$files = Invoke-DurableActivity -FunctionName 'E2_GetFileList' -Input $rootDirectory

# Fan-out: schedule parallel uploads for each file
$parallelTasks = @()
foreach ($file in $files) {
    $parallelTasks += Invoke-DurableActivity -FunctionName 'E2_CopyFileToBlob' -Input $file -NoWait
}

# Fan-in: wait for all uploads and sum the results
$results = Wait-DurableTask -Task $parallelTasks
$totalBytes = ($results | Measure-Object -Sum).Sum

$totalBytes

@FunctionName("E2_BackupSiteContent")
public long backupSiteContent(
        @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx) {

    String rootDirectory = ctx.getInput(String.class);

    // Get all files in the directory
    List<String> files = ctx.callActivity("E2_GetFileList", rootDirectory, List.class).await();

    // Fan-out: schedule parallel uploads for each file
    List<Task<Long>> parallelTasks = new ArrayList<>();
    for (String file : files) {
        parallelTasks.add(ctx.callActivity("E2_CopyFileToBlob", file, Long.class));
    }

    // Fan-in: wait for all uploads and sum the results
    List<Long> results = ctx.allOf(parallelTasks).await();
    long totalBytes = 0;
    for (Long bytes : results) {
        totalBytes += bytes;
    }

    return totalBytes;
}

De enskilda anropen till E2_CopyFileToBlob inväntas inte var för sig, så de körs därför parallellt. När orkestratören skickar uppgiftslistan till ctx.allOf(parallelTasks), returnerar den en uppgift som inte slutförs förrän alla kopieringsoperationer är klara. När alla uppgifter är klara summerar orkestratorn resultaten för att ladda upp totala bytes.

using Microsoft.DurableTask;
using System.Collections.Generic;
using System.Threading.Tasks;

[DurableTask]
public class ParallelProcessingOrchestration : TaskOrchestrator<List<string>, Dictionary<string, int>>
{
    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
        var 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;
    }
}

Använd Task.WhenAll() för att vänta tills alla parallella uppgifter har slutförts. Durable Task SDK säkerställer att uppgifterna kan köras på flera datorer samtidigt och att körningen är motståndskraftig mot omstarter av processer.

import {
  OrchestrationContext,
  TOrchestrator,
  whenAll,
} from "@microsoft/durabletask-js";

const fanOutFanInOrchestrator: TOrchestrator = async function* (
  ctx: OrchestrationContext,
  workItems: string[]
): any {
  // Fan-out: create a task for each work item in parallel
  const tasks = workItems.map((item) => ctx.callActivity(processWorkItem, item));

  // Wait for all parallel tasks to complete
  const results: number[] = yield whenAll(tasks);

  // Fan-in: aggregate all results
  const aggregatedResult = yield ctx.callActivity(aggregateResults, results);

  return aggregatedResult;
};

Använd whenAll() för att vänta tills alla parallella uppgifter har slutförts. Durable Task SDK säkerställer att uppgifterna kan köras på flera datorer samtidigt och att körningen är motståndskraftig mot omstarter av processer.

from durabletask import task

def fan_out_fan_in_orchestrator(ctx: task.OrchestrationContext, work_items: list) -> dict:
    """Orchestrator demonstrating fan-out/fan-in pattern."""

    # 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
    results = yield task.when_all(parallel_tasks)

    # Fan-in: Aggregate all the results
    final_result = yield ctx.call_activity(aggregate_results, input=results)

    return final_result

Använd task.when_all() för att vänta tills alla parallella uppgifter har slutförts. Durable Task SDK säkerställer att uppgifterna kan köras på flera datorer samtidigt och att körningen är motståndskraftig mot omstarter av processer.

Det här exemplet är tillgängligt för .NET, JavaScript, Java och Python.

import com.microsoft.durabletask.*;
import java.util.List;
import java.util.stream.Collectors;

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

                // Fan-out: Create a task for each input item
                List<Task<Integer>> tasks = inputs.stream()
                    .map(input -> ctx.callActivity("CountWords", input.toString(), Integer.class))
                    .collect(Collectors.toList());

                // Wait for all parallel tasks to complete
                List<Integer> allResults = ctx.allOf(tasks).await();

                // Fan-in: Aggregate results
                int totalCount = allResults.stream().mapToInt(Integer::intValue).sum();

                ctx.complete(totalCount);
            };
        }
    })
    .build();

Använd ctx.allOf(tasks).await() för att vänta tills alla parallella uppgifter har slutförts. Durable Task SDK säkerställer att uppgifterna kan köras på flera datorer samtidigt och att körningen är motståndskraftig mot omstarter av processer.

Filen function.json för E2_GetFileList ser ut som i följande exempel:

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "rootDirectory",
      "type": "activityTrigger",
      "direction": "in"
    }
  ]
}

Här är implementeringen:

import os
from os.path import dirname
from typing import List

def main(rootDirectory: str) -> List[str]:

    all_file_paths = []
    # We walk the file system
    for path, _, files in os.walk(rootDirectory):
        # We copy the code for activities and orchestrators
        if "E2_" in path:
            # For each file, we add their full-path to the list
            for name in files:
                if name == "__init__.py" or name == "function.json":
                    file_path = os.path.join(path, name)
                    all_file_paths.append(file_path)
    
    return all_file_paths

Aktiviteten E2_GetFileList samlar rekursivt in filvägar från den angivna katalogen:

param($rootDirectory)

Get-ChildItem -Path $rootDirectory -Recurse -File | Select-Object -ExpandProperty FullName

@FunctionName("E2_GetFileList")
public List<String> getFileList(
        @DurableActivityTrigger(name = "rootDirectory") String rootDirectory) {

    File root = new File(rootDirectory);
    List<String> files = new ArrayList<>();
    collectFiles(root, files);
    return files;
}

private void collectFiles(File directory, List<String> files) {
    File[] entries = directory.listFiles();
    if (entries != null) {
        for (File entry : entries) {
            if (entry.isDirectory()) {
                collectFiles(entry, files);
            } else {
                files.add(entry.getAbsolutePath());
            }
        }
    }
}

Den function.json filen för E2_CopyFileToBlob är lika enkel:

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "filePath",
      "type": "activityTrigger",
      "direction": "in"
    }
  ]
}

Python-implementeringen använder Azure Storage SDK för Python för att ladda upp filerna till Azure Blob Storage.

import os
import pathlib
from azure.storage.blob import BlobServiceClient
from azure.core.exceptions import ResourceExistsError

connect_str = os.getenv('AzureWebJobsStorage')

def main(filePath: str) -> str:
    # Create the BlobServiceClient object which will be used to create a container client
    blob_service_client = BlobServiceClient.from_connection_string(connect_str)
    
    # Create a unique name for the container
    container_name = "backups"
    
    # Create the container if it does not exist
    try:
        blob_service_client.create_container(container_name)
    except ResourceExistsError:
        pass

    # Create a blob client using the local file name as the name for the blob
    parent_dir, fname = pathlib.Path(filePath).parts[-2:] # Get last two path components
    blob_name = parent_dir + "_" + fname
    blob_client = blob_service_client.get_blob_client(container=container_name, blob=blob_name)

    # Count bytes in file
    byte_count = os.path.getsize(filePath)

    # Upload the created file
    with open(filePath, "rb") as data:
        blob_client.upload_blob(data)

    return byte_count

Aktiviteten E2_CopyFileToBlob läser en fil och laddar upp den till Azure Blob Storage:

param($filePath)

$storageContext = New-AzStorageContext -ConnectionString $env:AzureWebJobsStorage
$container = "backups"

# Create the container if it doesn't exist
New-AzStorageContainer -Name $container -Context $storageContext -ErrorAction SilentlyContinue

$blobName = $filePath.Substring([System.IO.Path]::GetPathRoot($filePath).Length).Replace('\', '/')
Set-AzStorageBlobContent -File $filePath -Container $container -Blob $blobName -Context $storageContext -Force

(Get-Item $filePath).Length

@FunctionName("E2_CopyFileToBlob")
public long copyFileToBlob(
        @DurableActivityTrigger(name = "filePath") String filePath) throws Exception {

    File file = new File(filePath);
    long byteCount = file.length();

    String blobPath = filePath
            .substring(File.listRoots()[0].getPath().length())
            .replace('\\', '/');

    String connectionString = System.getenv("AzureWebJobsStorage");
    BlobContainerClient containerClient = new BlobContainerClientBuilder()
            .connectionString(connectionString)
            .containerName("backups")
            .buildClient();
    containerClient.createIfNotExists();

    BlobClient blobClient = containerClient.getBlobClient(blobPath);
    blobClient.uploadFromFile(filePath, true);

    return byteCount;
}

using Microsoft.DurableTask;
using Microsoft.Extensions.Logging;
using System.Collections.Generic;
using System.Threading.Tasks;

[DurableTask]
public class ProcessWorkItemActivity : TaskActivity<string, Dictionary<string, int>>
{
    private readonly ILogger<ProcessWorkItemActivity> _logger;

    public ProcessWorkItemActivity(ILogger<ProcessWorkItemActivity> logger)
    {
        _logger = logger;
    }

    public override Task<Dictionary<string, int>> RunAsync(TaskActivityContext context, string workItem)
    {
        _logger.LogInformation("Processing work item: {WorkItem}", workItem);

        // Process the work item (where you do the actual work)
        var result = new Dictionary<string, int>
        {
            { workItem, workItem.Length }
        };

        return Task.FromResult(result);
    }
}

import { ActivityContext } from "@microsoft/durabletask-js";

const processWorkItem = async (
  _ctx: ActivityContext,
  item: string
): Promise<number> => {
  console.log(`Processing work item: "${item}"`);
  return item.length;
};

Till skillnad från orkestrerare kan aktiviteter utföra I/O-åtgärder som HTTP-anrop, databasfrågor och filåtkomst.

from durabletask import task

def process_work_item(ctx: task.ActivityContext, item: int) -> dict:
    """Activity processing a single work item."""
    # Process the work item (where you do the actual work)
    result = item * item
    return {"item": item, "result": result}

Det här exemplet visas för .NET, JavaScript, Java och Python.

import java.util.StringTokenizer;

// Activity registration
.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();
        };
    }
})

using Microsoft.DurableTask;
using Microsoft.Extensions.Logging;
using System.Collections.Generic;
using System.Threading.Tasks;

[DurableTask]
public class AggregateResultsActivity : TaskActivity<Dictionary<string, int>[], Dictionary<string, int>>
{
    private readonly ILogger<AggregateResultsActivity> _logger;

    public AggregateResultsActivity(ILogger<AggregateResultsActivity> logger)
    {
        _logger = logger;
    }

    public override Task<Dictionary<string, int>> RunAsync(
        TaskActivityContext context, Dictionary<string, int>[] results)
    {
        _logger.LogInformation("Aggregating {Count} results", results.Length);

        // Combine all results into one aggregated result
        var aggregatedResult = new Dictionary<string, int>();

        foreach (var result in results)
        {
            foreach (var kvp in result)
            {
                aggregatedResult[kvp.Key] = kvp.Value;
            }
        }

        return Task.FromResult(aggregatedResult);
    }
}

import { ActivityContext } from "@microsoft/durabletask-js";

const aggregateResults = async (
  _ctx: ActivityContext,
  results: number[]
): Promise<object> => {
  const total = results.reduce((sum, val) => sum + val, 0);
  return {
    totalItems: results.length,
    sum: total,
    average: results.length > 0 ? total / results.length : 0,
  };
};

Till skillnad från orkestrerare kan aktiviteter utföra I/O-åtgärder som HTTP-anrop, databasfrågor och filåtkomst.

from durabletask import task

def aggregate_results(ctx: task.ActivityContext, results: list) -> dict:
    """Activity aggregating results from multiple work items."""
    sum_result = sum(item["result"] for item in results)
    return {
        "total_items": len(results),
        "sum": sum_result,
        "average": sum_result / len(results) if results else 0
    }

Det här exemplet visas för .NET, JavaScript, Java och Python.

I det Java exemplet aggregerar orkestreraren resultat när ctx.allOf(tasks).await() returnerar.

// Schedule the orchestration with a list of work items
var workItems = new List<string> { "item1", "item2", "item3", "item4", "item5" };
string instanceId = await client.ScheduleNewOrchestrationInstanceAsync(
    nameof(ParallelProcessingOrchestration), workItems);

// Wait for completion
var result = await client.WaitForInstanceCompletionAsync(instanceId, getInputsAndOutputs: true);
Console.WriteLine($"Result: {result.ReadOutputAs<Dictionary<string, int>>().Count} items processed");

import {
  DurableTaskAzureManagedClientBuilder,
} from "@microsoft/durabletask-js-azuremanaged";

const connectionString =
  process.env.DURABLE_TASK_SCHEDULER_CONNECTION_STRING ||
  "Endpoint=http://localhost:8080;Authentication=None;TaskHub=default";

const client = new DurableTaskAzureManagedClientBuilder()
  .connectionString(connectionString)
  .build();

const workItems = ["item1", "item2", "item3", "item4", "item5"];
const instanceId = await client.scheduleNewOrchestration(fanOutFanInOrchestrator, workItems);
const state = await client.waitForOrchestrationCompletion(instanceId, true, 30);
console.log(`Result: ${state?.serializedOutput}`);

Skapa DurableTaskAzureManagedClientBuilder med hjälp av en anslutningssträng för Durable Task Scheduler. Använd scheduleNewOrchestration för att starta en orkestrering och använd waitForOrchestrationCompletion för att vänta tills den har slutförts.

# Schedule the orchestration with a list of work items
work_items = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
instance_id = client.schedule_new_orchestration(fan_out_fan_in_orchestrator, input=work_items)

# Wait for completion
result = client.wait_for_orchestration_completion(instance_id, timeout=60)
print(f"Result: {result.serialized_output}")

Det här exemplet visas för .NET, JavaScript, Java och Python.

import java.time.Duration;
import java.util.Arrays;
import java.util.List;

// Schedule the orchestration with a list of strings
List<String> sentences = Arrays.asList(
    "Hello, world!",
    "The quick brown fox jumps over the lazy dog.",
    "Always remember you are absolutely unique.");

String instanceId = client.scheduleNewOrchestrationInstance(
    "FanOutFanIn_WordCount",
    new NewOrchestrationInstanceOptions().setInput(sentences));

// Wait for completion
OrchestrationMetadata result = client.waitForInstanceCompletion(instanceId, Duration.ofSeconds(30), true);
System.out.println("Total word count: " + result.readOutputAs(int.class));