Wzorzec wentylatora/wentylatora

• Ukończ artykuł Szybki start
• Sklonuj lub pobierz projekt przykładowy z GitHub
  • Izolowane próbki modeli
  • Próbki modeli w trakcie przetwarzania

• Ukończ artykuł Szybki start
• Skopiuj lub pobierz przykładowy projekt z GitHub

• Ukończ artykuł Szybki start
• Skopiuj lub pobierz przykładowy projekt z GitHub

• Ukończ artykuł Szybki start

Durable Functions nie obsługuje Go. Aby zobaczyć przykłady Go, wybierz Durable Task SDK na górze tego artykułu.

Poniższy kod demonstruje implementację funkcji orkiestratora:

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

Funkcja używa standardowego function.json dla funkcji orkiestratora.

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

Poniższy kod demonstruje implementację funkcji orkiestratora:

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)

Zwróć uwagę na yield context.task_all(tasks); wiersz. Kod nie zwraca poszczególnych wywołań do E2_CopyFileToBlob, więc są one uruchamiane równolegle. Kiedy orchestrator przekazuje tablicę zadań do context.task_all, zwraca zadanie, które nie zostanie ukończone, dopóki wszystkie operacje kopiowania się nie zakończą. Jeśli znasz asyncio.gather język Python, ta koncepcja nie jest dla Ciebie nowa. Dzięki rozszerzeniu Durable Functions te zadania są uruchamiane jednocześnie na wielu maszynach wirtualnych, a kompleksowe wykonywanie jest odporne na odtwarzanie procesów.

Gdy orkiestrator zwróci context.task_all, wszystkie wywołania funkcji zostaną zakończone i zwrócą wartości. Każde wywołanie zwraca E2_CopyFileToBlob liczbę przesłanych bajtów, więc możesz obliczyć sumę całkowitej liczby bajtów, sumując wszystkie wartości zwracane.

Orkiestrator otrzymuje listę plików, a następnie rozdziela zadanie kopiowania każdego pliku równolegle do magazynu obiektów blob przy użyciu -NoWait. Po zakończeniu wszystkich równoległych zadań wyniki są sumowane.

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

Nie oczekuje się osobno na poszczególne wywołania do E2_CopyFileToBlob, więc są wykonywane równolegle. Gdy orkestrator przekazuje listę zadań do ctx.allOf(parallelTasks), zwraca zadanie, które nie kończy się, dopóki wszystkie operacje kopiowania nie zostaną zakończone. Po zakończeniu wszystkich zadań orkestrator sumuje wyniki, aby przesłać łączną liczbę bajtów.

Durable Functions nie obsługuje Go. Aby zobaczyć przykłady Go, wybierz Durable Task SDK na górze tego artykułu.

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

Użyj Task.WhenAll(), aby poczekać na ukończenie wszystkich zadań równoległych. Zestaw Durable Task SDK zapewnia, że zadania mogą być uruchamiane na wielu maszynach jednocześnie, a wykonanie jest odporne na ponowne uruchomienie procesu.

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

Użyj whenAll(), aby poczekać na ukończenie wszystkich zadań równoległych. Zestaw Durable Task SDK zapewnia, że zadania mogą być uruchamiane na wielu maszynach jednocześnie, a wykonanie jest odporne na ponowne uruchomienie procesu.

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

Użyj task.when_all(), aby poczekać na ukończenie wszystkich zadań równoległych. Zestaw Durable Task SDK zapewnia, że zadania mogą być uruchamiane na wielu maszynach jednocześnie, a wykonanie jest odporne na ponowne uruchomienie procesu.

Ten przykład jest dostępny dla .NET, JavaScript, Java, Python i Go.

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

Użyj ctx.allOf(tasks).await(), aby poczekać na ukończenie wszystkich zadań równoległych. Zestaw Durable Task SDK zapewnia, że zadania mogą być uruchamiane na wielu maszynach jednocześnie, a wykonanie jest odporne na ponowne uruchomienie procesu.

Zarejestruj ten orkestrator jako FanOutFanIn. Sekcja Aktywności definiuje limit , typy i oraz aktywności i . Rejestruj aktywności jako ProcessWorkItem i AggregateResults.

package main

import (
	"fmt"

	"github.com/microsoft/durabletask-go/task"
)

func fanOutFanIn(ctx *task.OrchestrationContext) (any, error) {
	var items []int64
	if err := ctx.GetInput(&items); err != nil {
		return nil, err
	}
	if len(items) > maxItems {
		return nil, fmt.Errorf("batch contains more than %d items", maxItems)
	}
	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)
	}
	results := make([]workResult, len(pending))
	for i, work := range pending {
		if err := work.Await(&results[i]); err != nil {
			return nil, err
		}
	}
	var result summary
	if err := ctx.CallActivity("AggregateResults", task.WithActivityInput(results)).Await(&result); err != nil {
		return nil, err
	}
	return result, nil
}

Zaplanuj każdą aktywność, zanim zaczekasz. WhenAll czeka na wszystkie zadania, w tym na zadania równorzędne, gdy jedno z nich zakończy się niepowodzeniem; Await dekoduje każdy wynik. Nie wdrażaj tego wzorca za pomocą natywnych goroutines wewnątrz orkiestratora.

Plik function.json pliku E2_GetFileList wygląda podobnie do następującego przykładu:

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

Oto implementacja:

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

E2_GetFileList Aktywność rekurencyjnie zbiera ścieżki plików z określonego katalogu:

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

Durable Functions nie obsługuje Go. Aby zobaczyć przykłady Go, wybierz Durable Task SDK na górze tego artykułu.

Plik function.json dla E2_CopyFileToBlob jest podobnie prosty.

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

Implementacja Python używa zestawu SDK Azure Storage dla Python aby przekazać pliki do 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

Aktywność E2_CopyFileToBlob odczytuje plik i przesyła go do 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;
}

Durable Functions nie obsługuje Go. Aby zobaczyć przykłady Go, wybierz Durable Task SDK na górze tego artykułu.

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

W przeciwieństwie do orkiestratorów działania mogą wykonywać operacje we/wy, takie jak wywołania HTTP, zapytania bazy danych i dostęp do plików.

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}

Ten przykład jest pokazany dla .NET, JavaScript, Java, Python i Go.

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

Zapisz następujące aktywności i typy w osobnym pliku w tym samym pakiecie:

package main

import (
	"fmt"

	"github.com/microsoft/durabletask-go/task"
)

const (
	maxItems     = 100
	maxMagnitude = 1_000_000
)

type workResult struct {
	Item   int64 `json:"item"`
	Result int64 `json:"result"`
}

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

func square(item int64) (workResult, error) {
	if item < -maxMagnitude || item > maxMagnitude {
		return workResult{}, fmt.Errorf("item %d exceeds the sample's safe arithmetic range", item)
	}
	return workResult{Item: item, Result: item * item}, nil
}

func processWorkItem(ctx task.ActivityContext) (any, error) {
	var item int64
	if err := ctx.GetInput(&item); err != nil {
		return nil, err
	}
	return square(item)
}

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

W przeciwieństwie do orkiestratorów działania mogą wykonywać operacje we/wy, takie jak wywołania HTTP, zapytania bazy danych i dostęp do plików.

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
    }

Ten przykład jest pokazany dla .NET, JavaScript, Java, Python i Go.

W przykładzie Java orkiestrator agreguje wyniki po tym, jak ctx.allOf(tasks).await() zakończy działanie.

Zapisz tę aktywność w innym pliku w tym samym pakiecie. Weryfikuje wyniki przed ich agregacją i obsługuje pustą partię bez dzielenia przez zero.

package main

import (
	"fmt"

	"github.com/microsoft/durabletask-go/task"
)

func aggregateResults(ctx task.ActivityContext) (any, error) {
	var results []workResult
	if err := ctx.GetInput(&results); err != nil {
		return nil, err
	}
	if len(results) > maxItems {
		return nil, fmt.Errorf("batch contains more than %d items", maxItems)
	}
	result := summary{TotalItems: len(results)}
	for _, item := range results {
		expected, err := square(item.Item)
		if err != nil {
			return nil, err
		}
		if item != expected {
			return nil, fmt.Errorf("incorrect square for item %d: %d", item.Item, item.Result)
		}
		result.Sum += item.Result
	}
	if result.TotalItems != 0 {
		result.Average = float64(result.Sum) / float64(result.TotalItems)
	}
	return result, nil
}

Aby uzyskać pełną aplikację z konfiguracją połączenia, zobacz szybki start Go.

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

Utwórz DurableTaskAzureManagedClientBuilder przy użyciu ciągu połączenia do trwałego harmonogramu zadań. Użyj scheduleNewOrchestration, aby rozpocząć orkiestrację, i użyj waitForOrchestrationCompletion, aby poczekać na zakończenie.

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

Ten przykład jest pokazany dla .NET, JavaScript, Java, Python i Go.

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

Aby uzyskać kompletny program, który uruchamia pracownika, planuje orkiestrację fan-out/fan-in i odczytuje wynik, zobacz quickstart Go. Więcej wzorców równoległych znajdziesz w Go SDK parallel sample.