Sub-orquestaciones

Las funciones de orquestación pueden llamar a otras funciones de orquestación como suborquestaciones. Una suborquestación se ejecuta como una entidad secundaria de la orquestación que la invoca (orquestación principal) y se comporta como una actividad desde la perspectiva del llamador: puede devolver un valor y generar excepciones que son capturadas por la orquestación principal y admitir el reintento automático.

Cuándo usar suborquestaciones

Utilice suborquestaciones cuando necesite:

• Crear bloques de creación de flujos de trabajo reutilizables: Extraiga un flujo de trabajo de varios pasos en su propio orquestador para que varias orquestaciones primarias puedan llamarla.
• Distribución ramificada de orquestaciones en paralelo: Programe muchas instancias del mismo orquestador simultáneamente y espere a que finalicen todas ellas.
• Organizar flujos de trabajo complejos: Divida una orquestación grande en partes con nombre y que se puedan probar en lugar de una sola función larga.

Nota:

Las suborquestaciones deben definirse en la misma aplicación que la orquestación primaria. Para llamar a orquestaciones en otra aplicación, use el patrón de sondeo HTTP 202 en su lugar. Para obtener más información, consulte Características HTTP.

En este artículo:

• Definición de una suborquestación — ejemplo de provisionamiento de un único dispositivo
• Ejecución de suborquestaciones en paralelo : patrón de distribución ramificada con identificadores de instancia deterministas

Nota:

En PowerShell, las sub-orquestaciones solo se admiten en el SDK independiente: AzureFunctions.PowerShell.Durable.SDK. Para conocer las diferencias entre el SDK independiente y el SDK integrado heredado, consulte la guía de migración.

Definición de una suborquestación

En el ejemplo siguiente se muestra un escenario de IoT en el que es necesario configurar varios dispositivos. La función representa el flujo de trabajo de instalación que se ejecuta para cada dispositivo:

C#

Modelo de trabajo aislado

public static async Task DeviceProvisioningOrchestration(
    [OrchestrationTrigger] TaskOrchestrationContext context, string deviceId)
{
    // Step 1: Create an installation package in blob storage and return a SAS URL.
    Uri sasUrl = await context.CallActivityAsync<Uri>("CreateInstallationPackage", deviceId);

    // Step 2: Notify the device that the installation package is ready.
    await context.CallActivityAsync("SendPackageUrlToDevice", (deviceId, sasUrl));

    // Step 3: Wait for the device to acknowledge that it has downloaded the new package.
    await context.WaitForExternalEvent<bool>("DownloadCompletedAck");

    // Step 4: ...
}

Modelo en proceso

public static async Task DeviceProvisioningOrchestration(
    [OrchestrationTrigger] IDurableOrchestrationContext context)
{
    string deviceId = context.GetInput<string>();

    // Step 1: Create an installation package in blob storage and return a SAS URL.
    Uri sasUrl = await context.CallActivityAsync<Uri>("CreateInstallationPackage", deviceId);

    // Step 2: Notify the device that the installation package is ready.
    await context.CallActivityAsync("SendPackageUrlToDevice", Tuple.Create(deviceId, sasUrl));

    // Step 3: Wait for the device to acknowledge that it has downloaded the new package.
    await context.WaitForExternalEvent<bool>("DownloadCompletedAck");

    // Step 4: ...
}

JavaScript

const df = require("durable-functions");

df.app.orchestration("deviceProvisioningOrchestration", function* (context) {
    const deviceId = context.df.getInput();

    // Step 1: Create an installation package in blob storage and return a SAS URL.
    const sasUrl = yield context.df.callActivity("createInstallationPackage", deviceId);

    // Step 2: Notify the device that the installation package is ready.
    yield context.df.callActivity("sendPackageUrlToDevice", { id: deviceId, url: sasUrl });

    // Step 3: Wait for the device to acknowledge that it has downloaded the new package.
    yield context.df.waitForExternalEvent("downloadCompletedAck");

    // Step 4: ...
});

Python

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

def orchestrator_function(context: df.DurableOrchestrationContext):
    device_id = context.get_input()

    # Step 1: Create an installation package in blob storage and return a SAS URL.
    sas_url = yield context.call_activity("CreateInstallationPackage", device_id)

    # Step 2: Notify the device that the installation package is ready.
    yield context.call_activity("SendPackageUrlToDevice", { "id": device_id, "url": sas_url })

    # Step 3: Wait for the device to acknowledge that it has downloaded the new package.
    yield context.call_activity("DownloadCompletedAck")

    # Step 4: ...

PowerShell

param($Context)

$deviceId = $Context.Input

# Step 1: Create an installation package in blob storage and return a SAS URL.
$sasUrl = Invoke-DurableActivity -FunctionName "CreateInstallationPackage" -Input $deviceId

# Step 2: Notify the device that the installation package is ready.
$deviceInfo = @{
    id = $deviceId
    url = $sasUrl
}
Invoke-DurableActivity -FunctionName "SendPackageUrlToDevice" -Input $deviceInfo

# Step 3: Wait for the device to acknowledge that it has downloaded the new package.
Start-DurableExternalEventListener -EventName "DownloadCompletedAck"

# Step 4: ...

Java

@FunctionName("DeviceProvisioningOrchestration")
public void deviceProvisioningOrchestration(
        @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx) {
    // Step 1: Create an installation package in blob storage and return a SAS URL.
    String deviceId = ctx.getInput(String.class);
    String blobUri = ctx.callActivity("CreateInstallPackage", deviceId, String.class).await();

    // Step 2: Notify the device that the installation package is ready.
    String[] args = { deviceId, blobUri };
    ctx.callActivity("SendPackageUrlToDevice", args).await();

    // Step 3: Wait for the device to acknowledge that it has downloaded the new package.
    ctx.waitForExternalEvent("DownloadCompletedAck").await();

    // Step 4: ...
}

C#

using Microsoft.DurableTask;

[DurableTask]
public class DeviceProvisioningOrchestration : TaskOrchestrator<string, object?>
{
    public override async Task<object?> RunAsync(TaskOrchestrationContext context, string deviceId)
    {
        // Step 1: Create an installation package in blob storage and return a SAS URL.
        Uri sasUrl = await context.CallActivityAsync<Uri>("CreateInstallationPackage", deviceId);

        // Step 2: Notify the device that the installation package is ready.
        await context.CallActivityAsync("SendPackageUrlToDevice", (deviceId, sasUrl.ToString()));

        // Step 3: Wait for the device to acknowledge that it has downloaded the new package.
        await context.WaitForExternalEvent<bool>("DownloadCompletedAck");

        // Step 4: ...
        return null;
    }
}

JavaScript

Este ejemplo se muestra para .NET, Java y Python.

Python

from durabletask import task

def device_provisioning_orchestrator(ctx: task.OrchestrationContext, device_id: str):
    # Step 1: Create an installation package in blob storage and return a SAS URL.
    sas_url = yield ctx.call_activity("create_installation_package", input=device_id)

    # Step 2: Notify the device that the installation package is ready.
    yield ctx.call_activity("send_package_url_to_device", input={"id": device_id, "url": sas_url})

    # Step 3: Wait for the device to acknowledge that it has downloaded the new package.
    yield ctx.wait_for_external_event("DownloadCompletedAck")

    # Step 4: ...

PowerShell

Este ejemplo se muestra para .NET, Java y Python.

Java

import com.microsoft.durabletask.TaskOrchestration;
import com.microsoft.durabletask.TaskOrchestrationContext;

public class DeviceProvisioningOrchestration implements TaskOrchestration {
    @Override
    public void run(TaskOrchestrationContext ctx) {
        String deviceId = ctx.getInput(String.class);

        // Step 1: Create an installation package in blob storage and return a SAS URL.
        String blobUri = ctx.callActivity("CreateInstallPackage", deviceId, String.class).await();

        // Step 2: Notify the device that the installation package is ready.
        String[] args = { deviceId, blobUri };
        ctx.callActivity("SendPackageUrlToDevice", args).await();

        // Step 3: Wait for the device to acknowledge that it has downloaded the new package.
        ctx.waitForExternalEvent("DownloadCompletedAck").await();

        // Step 4: ...
    }
}

Esta función de orquestador puede ejecutarse de forma independiente para la configuración de un dispositivo único o un orquestador primario puede programarla como una sub-orquestación mediante la API call-sub-orchestrator .

Ejecución de suborquestaciones en paralelo

En el siguiente ejemplo se muestra un orquestador primario que distribuye de manera ramificada varias suborquestaciones en paralelo. Algunos lenguajes usan un identificador de instancia secundario determinista (derivado del identificador de instancia del elemento más un índice), para evitar las suborquestaciones duplicadas durante la repetición.

C#

Modelo de trabajo aislado

[Function("ProvisionNewDevices")]
public static async Task ProvisionNewDevices(
    [OrchestrationTrigger] TaskOrchestrationContext context)
{
    string[] deviceIds = await context.CallActivityAsync<string[]>("GetNewDeviceIds");

    // Run multiple device provisioning flows in parallel
    var provisioningTasks = new List<Task>();
    foreach (string deviceId in deviceIds)
    {
        Task provisionTask = context.CallSubOrchestratorAsync("DeviceProvisioningOrchestration", deviceId);
        provisioningTasks.Add(provisionTask);
    }

    await Task.WhenAll(provisioningTasks);

    // ...
}

Modelo en proceso

[FunctionName("ProvisionNewDevices")]
public static async Task ProvisionNewDevices(
    [OrchestrationTrigger] IDurableOrchestrationContext context)
{
    string[] deviceIds = await context.CallActivityAsync<string[]>("GetNewDeviceIds");

    // Run multiple device provisioning flows in parallel
    var provisioningTasks = new List<Task>();
    foreach (string deviceId in deviceIds)
    {
        Task provisionTask = context.CallSubOrchestratorAsync("DeviceProvisioningOrchestration", deviceId);
        provisioningTasks.Add(provisionTask);
    }

    await Task.WhenAll(provisioningTasks);

    // ...
}

JavaScript

const df = require("durable-functions");

df.app.orchestration("provisionNewDevices", function* (context) {
    const deviceIds = yield context.df.callActivity("getNewDeviceIds");

    // Run multiple device provisioning flows in parallel
    const provisioningTasks = [];
    var id = 0;
    for (const deviceId of deviceIds) {
        const child_id = context.df.instanceId + `:${id}`;
        const provisionTask = context.df.callSubOrchestrator(
            "deviceProvisioningOrchestration",
            deviceId,
            child_id
        );
        provisioningTasks.push(provisionTask);
        id++;
    }

    yield context.df.Task.all(provisioningTasks);

    // ...
});

Python

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

def orchestrator_function(context: df.DurableOrchestrationContext):

    device_IDs = yield context.call_activity("GetNewDeviceIds")

    # Run multiple device provisioning flows in parallel
    provisioning_tasks = []
    id_ = 0
    for device_id in device_IDs:
        child_id = f"{context.instance_id}:{id_}"
        provision_task = context.call_sub_orchestrator("DeviceProvisioningOrchestration", device_id, child_id)
        provisioning_tasks.append(provision_task)
        id_ += 1

    yield context.task_all(provisioning_tasks)

    # ...

PowerShell

param($Context)

$deviceIds = Invoke-DurableActivity -FunctionName "GetNewDeviceIds"

# Run multiple device setting up flows in parallel
$provisioningTasks = @()
for ($i = 0; $i -lt $deviceIds.Count; $i++) {
    $deviceId = $deviceIds[$i]
    $childId = "$($Context.InstanceId):$i"
    $provisionTask = Invoke-DurableSubOrchestrator `
        -FunctionName "DeviceProvisioningOrchestration" `
        -Input $deviceId `
        -InstanceId $childId `
        -NoWait
    $provisioningTasks += $provisionTask
}

Wait-DurableTask -Task $provisioningTasks

# ...

Java

@FunctionName("ProvisionNewDevices")
public void provisionNewDevices(
        @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx) {
    List<?> deviceIDs = ctx.getInput(List.class);

    // Schedule each device provisioning sub-orchestration to run in parallel
    List<Task<Void>> parallelTasks = deviceIDs.stream()
        .map(device -> ctx.callSubOrchestrator("DeviceProvisioningOrchestration", device))
        .collect(Collectors.toList());

    // ...
}

Pasos siguientes

Obtenga información sobre cómo establecer un estado de orquestación personalizado

• Control de errores y compensación
• Introducción a las orquestaciones
• Introducción al Programador de tareas durables

C#

using Microsoft.DurableTask;

[DurableTask]
public class ProvisionNewDevices : TaskOrchestrator<object?, object?>
{
    public override async Task<object?> RunAsync(TaskOrchestrationContext context, object? input)
    {
        string[] deviceIds = await context.CallActivityAsync<string[]>("GetNewDeviceIds");

        // Run multiple device provisioning flows in parallel
        var provisioningTasks = new List<Task>();
        foreach (string deviceId in deviceIds)
        {
            Task provisionTask = context.CallSubOrchestratorAsync("DeviceProvisioningOrchestration", deviceId);
            provisioningTasks.Add(provisionTask);
        }

        await Task.WhenAll(provisioningTasks);
        return null;
    }
}

JavaScript

Este ejemplo se muestra para .NET, Java y Python.

Python

from durabletask import task

def provision_new_devices(ctx: task.OrchestrationContext, _):
    device_ids = yield ctx.call_activity("get_new_device_ids")

    # Run multiple device provisioning flows in parallel
    provisioning_tasks = []
    for device_id in device_ids:
        provision_task = ctx.call_sub_orchestrator(device_provisioning_orchestrator, input=device_id)
        provisioning_tasks.append(provision_task)

    yield task.when_all(provisioning_tasks)

PowerShell

Este ejemplo se muestra para .NET, Java y Python.

Java

import com.microsoft.durabletask.TaskOrchestration;
import com.microsoft.durabletask.TaskOrchestrationContext;

public class ProvisionNewDevices implements TaskOrchestration {
    @Override
    public void run(TaskOrchestrationContext ctx) {
        List<?> deviceIDs = ctx.getInput(List.class);

        // Schedule each device provisioning sub-orchestration to run in parallel
        List<Task<Void>> parallelTasks = deviceIDs.stream()
            .map(device -> ctx.callSubOrchestrator("DeviceProvisioningOrchestration", device))
            .collect(Collectors.toList());

        ctx.allOf(parallelTasks).await();
    }
}

Pasos siguientes

Introducción a los SDK de Durable Task

• Control de errores y compensación
• Introducción a las orquestaciones
• Introducción al Programador de tareas durables