Padrão fan-out/fan-in

• Conclua o artigo de início rápido
• Clonar ou baixar o projeto de exemplos do GitHub (usa o modelo em processo)

• Conclua o artigo de início rápido
• Clone ou baixe o projeto de amostras do GitHub

• Conclua o artigo de início rápido
• Clone ou baixe o projeto de amostras do GitHub

• Conclua o artigo de início rápido

O código a seguir demonstra a implementação da função de orquestrador:

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

A função usa o function.json padrão para funções de orquestrador.

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

O código a seguir demonstra a implementação da função de orquestrador:

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)

Observe a linha yield context.task_all(tasks);. O código não gera as chamadas individuais para E2_CopyFileToBlob, portanto, elas são executadas em paralelo. Quando o orquestrador passa a matriz de tarefas para context.task_all, ele retorna uma tarefa que só é concluída quando todas as operações de cópia são concluídas. Se você estiver familiarizado com asyncio.gather o Python, esse conceito não é novo para você. Com a extensão Durable Functions, essas tarefas são executadas em várias máquinas virtuais simultaneamente e a execução de ponta a ponta é resiliente à reciclagem de processos.

Depois que o orquestrador produzir context.task_all, todas as chamadas de função serão concluídas e retornarão valores. Cada chamada para E2_CopyFileToBlob retorna o número de bytes carregados, para que possamos calcular a contagem total de bytes somando todos os valores retornados juntos.

Um exemplo do PowerShell ainda não está disponível.

Um exemplo de Java ainda não está disponível.

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

Use Task.WhenAll() para aguardar a conclusão de todas as tarefas paralelas. O SDK de Tarefas Duráveis garante que as tarefas possam ser executadas em vários computadores simultaneamente e que a execução seja resiliente a reinicializações de processos.

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

Use whenAll() para aguardar a conclusão de todas as tarefas paralelas. O SDK de Tarefas Duráveis garante que as tarefas possam ser executadas em vários computadores simultaneamente e que a execução seja resiliente a reinicializações de processos.

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

Use task.when_all() para aguardar a conclusão de todas as tarefas paralelas. O SDK de Tarefas Duráveis garante que as tarefas possam ser executadas em vários computadores simultaneamente e que a execução seja resiliente a reinicializações de processos.

Este exemplo está disponível para .NET, JavaScript, Java e 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();

Use ctx.allOf(tasks).await() para aguardar a conclusão de todas as tarefas paralelas. O SDK de Tarefas Duráveis garante que as tarefas possam ser executadas em vários computadores simultaneamente e que a execução seja resiliente a reinicializações de processos.

O arquivo function.json para E2_GetFileList está conforme o exemplo a seguir:

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

Esta é a implementação:

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

Exemplo do PowerShell em breve.

Exemplo de Java em breve.

O arquivo function.json para E2_CopyFileToBlob também é simples:

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

A implementação do Python usa o SDK Azure Storage para Python para carregar os arquivos para 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

Exemplo do PowerShell em breve.

Exemplo de Java em breve.

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

Ao contrário dos orquestradores, as atividades podem executar operações de E/S, como chamadas HTTP, consultas de banco de dados e acesso a arquivos.

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}

Este exemplo é mostrado para .NET, JavaScript, Java e 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,
  };
};

Ao contrário dos orquestradores, as atividades podem executar operações de E/S, como chamadas HTTP, consultas de banco de dados e acesso a arquivos.

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
    }

Este exemplo é mostrado para .NET, JavaScript, Java e Python.

No exemplo de Java, o orquestrador agrega resultados depois do retorno de ctx.allOf(tasks).await().

// 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}`);

Crie o DurableTaskAzureManagedClientBuilder usando uma cadeia de conexão para o Agendador de Tarefas Duráveis. Use scheduleNewOrchestration para iniciar uma orquestração e use waitForOrchestrationCompletion para aguardar a conclusão.

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

Este exemplo é mostrado para .NET, JavaScript, Java e 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));