Functiekoppeling

• Het quickstart-artikel voltooien
• Clone of download het voorbeeldproject van GitHub

• Het quickstart-artikel voltooien
• Clone of download het voorbeeldproject van GitHub

• Het quickstart-artikel voltooien
• Clone of download het voorbeeldproject van GitHub

• Het quickstart-artikel voltooien

[FunctionName("E1_HelloSequence")]
public static async Task<List<string>> Run(
    [OrchestrationTrigger] IDurableOrchestrationContext context)
{
    var outputs = new List<string>();

    outputs.Add(await context.CallActivityAsync<string>("E1_SayHello", "Tokyo"));
    outputs.Add(await context.CallActivityAsync<string>("E1_SayHello", "Seattle"));
    outputs.Add(await context.CallActivityAsync<string>("E1_SayHello_DirectInput", "London"));

    // returns ["Hello Tokyo!", "Hello Seattle!", "Hello London!"]
    return outputs;
}

Alle C#-indelingsfuncties moeten een parameter van het type DurableOrchestrationContext hebben, die aanwezig is in de assembly Microsoft.Azure.WebJobs.Extensions.DurableTask. Met dit contextobject kunt u andere activiteitsfuncties aanroepen en invoerparameters doorgeven met behulp van de CallActivityAsync methode.

De code roept E1_SayHello drie keer op volgorde aan met verschillende parameterwaarden. De retourwaarde van elke aanroep wordt toegevoegd aan de outputs lijst, die aan het einde van de functie wordt geretourneerd.

function.json

Als u Visual Studio Code of de Azure-portal gebruikt voor ontwikkeling, vindt u hier de inhoud van het bestand function.json voor de orchestratorfunctie. De meeste orchestrator-function.json bestanden zien er bijna precies zo uit.

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

Het belangrijkste is het orchestrationTrigger bindingstype. Alle orchestratorfuncties moeten dit triggertype gebruiken.

__init__.py

Dit is de orchestrator-functie:

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


def orchestrator_function(context: df.DurableOrchestrationContext):
    result1 = yield context.call_activity('E1_SayHello', "Tokyo")
    result2 = yield context.call_activity('E1_SayHello', "Seattle")
    result3 = yield context.call_activity('E1_SayHello', "London")
    return [result1, result2, result3]

main = df.Orchestrator.create(orchestrator_function)

Alle Python indelingsfuncties moeten het durable-functions-pakket bevatten. Het is een bibliotheek waarmee u Durable Functions kunt schrijven in Python. Twee belangrijke verschillen tussen een orchestratorfunctie en andere Python functies:

1. De orchestratorfunctie is een generatorfunctie.
2. Het bestand registreert de orchestratorfunctie door aan het einde van het bestand aan te geven main = df.Orchestrator.create(<orchestrator function name>) . Hiermee kunt u onderscheid maken tussen andere helperfuncties die in het bestand zijn gedeclareerd.

Met context het object kunt u andere activiteitsfuncties aanroepen en invoerparameters doorgeven met behulp van de call_activity methode. De code roept E1_SayHello drie keer achter elkaar aan met verschillende parameterwaarden. Met yield wordt aangegeven dat de uitvoering moet wachten tot de aanroepen van de asynchrone functie zijn geretourneerd. De retourwaarde van elke aanroep wordt aan het einde van de functie geretourneerd.

PowerShell-voorbeeld is nog niet beschikbaar.

Java voorbeeld is nog niet beschikbaar.

Deze code toont een orchestrator die drie activiteiten op volgorde aanroept en elke uitvoer doorgeeft aan de volgende activiteit:

using Microsoft.DurableTask;

[DurableTask]
public class GreetingOrchestration : TaskOrchestrator<string, string>
{
    public override async Task<string> RunAsync(TaskOrchestrationContext context, string name)
    {
        // Step 1: Say hello to the person
        string greeting = await context.CallActivityAsync<string>(nameof(SayHelloActivity), name);

        // Step 2: Process the greeting
        string processedGreeting = await context.CallActivityAsync<string>(nameof(ProcessGreetingActivity), greeting);

        // Step 3: Finalize the response
        string finalResponse = await context.CallActivityAsync<string>(nameof(FinalizeResponseActivity), processedGreeting);

        return finalResponse;
    }
}

Alle .NET orchestrators nemen over van TaskOrchestrator<TInput, TOutput>. Hiermee TaskOrchestrationContext kunt u activiteiten aanroepen met behulp van CallActivityAsync. De code roept drie activiteiten op volgorde aan, waarbij elke activiteit de uitvoer van de vorige activiteit ontvangt.

De volgende code toont een orchestrator die drie activiteiten op volgorde aanroept:

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

const greetingOrchestrator: TOrchestrator = async function* (
  ctx: OrchestrationContext,
  name: string
): any {
  // Step 1: Say hello to the person
  const greeting: string = yield ctx.callActivity(sayHello, name);

  // Step 2: Process the greeting
  const processedGreeting: string = yield ctx.callActivity(processGreeting, greeting);

  // Step 3: Finalize the response
  const finalResponse: string = yield ctx.callActivity(finalizeResponse, processedGreeting);

  return finalResponse;
};

Alle JavaScript orchestrators zijn generatorfuncties (async function*) die yield gebruiken om activiteiten aan te roepen. De methode van de orkestratiecontext callActivity schema de uitvoering van activiteiten. De code roept drie activiteiten op volgorde aan, waarbij de uitvoer van elke activiteit wordt doorgegeven aan de volgende.

De volgende code toont een orchestrator die drie activiteiten op volgorde aanroept:

from durabletask import task

# Orchestrator function
def function_chaining_orchestrator(ctx: task.OrchestrationContext, name: str) -> str:
    """Orchestrator that demonstrates function chaining pattern."""

    # Call first activity
    greeting = yield ctx.call_activity(say_hello, input=name)

    # Call second activity with the result from first activity
    processed_greeting = yield ctx.call_activity(process_greeting, input=greeting)

    # Call third activity with the result from second activity
    final_response = yield ctx.call_activity(finalize_response, input=processed_greeting)

    return final_response

Alle Python orchestrators zijn generatorfuncties die gebruikmaken van yield om activiteiten aan te roepen. De methode van de orkestratiecontext call_activity plant de uitvoering van activiteiten in. De code roept drie activiteiten op volgorde aan, waarbij de uitvoer van elke activiteit wordt doorgegeven aan de volgende.

Dit voorbeeld wordt weergegeven voor .NET, JavaScript, Java en Python.

De volgende code toont een orchestrator die drie activiteiten op volgorde aanroept:

import com.microsoft.durabletask.*;

DurableTaskGrpcWorker worker = DurableTaskSchedulerWorkerExtensions.createWorkerBuilder(connectionString)
    .addOrchestration(new TaskOrchestrationFactory() {
        @Override
        public String getName() { return "ActivityChaining"; }

        @Override
        public TaskOrchestration create() {
            return ctx -> {
                String input = ctx.getInput(String.class);

                // Call activities in sequence, passing output from one to the next
                String x = ctx.callActivity("Reverse", input, String.class).await();
                String y = ctx.callActivity("Capitalize", x, String.class).await();
                String z = ctx.callActivity("ReplaceWhitespace", y, String.class).await();

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

In Java worden orchestrators gedefinieerd met behulp van TaskOrchestrationFactory. De methode van callActivity de context plant de uitvoering van de activiteit en await() wacht op het resultaat. De code roept drie activiteiten op volgorde aan, waarbij de uitvoer van elke activiteit wordt doorgegeven aan de volgende.

[FunctionName("E1_SayHello")]
public static string SayHello([ActivityTrigger] IDurableActivityContext context)
{
    string name = context.GetInput<string>();
    return $"Hello {name}!";
}

Activiteiten gebruiken het ActivityTrigger kenmerk. Gebruiken IDurableActivityContext voor activiteitsacties, zoals het lezen van invoer met GetInput<T>.

E1_SayHello formatteert een begroetingsreeks.

In plaats van een binding op IDurableActivityContext toe te passen, bind rechtstreeks aan het type dat is doorgegeven aan de activiteitsfunctie. Voorbeeld:

[FunctionName("E1_SayHello_DirectInput")]
public static string SayHelloDirectInput([ActivityTrigger] string name)
{
    return $"Hello {name}!";
}

E1_SayHello/function.json

Het function.json-bestand voor de activiteitsfunctie E1_SayHello is vergelijkbaar met E1_HelloSequence het bestand, behalve dat het een activityTrigger bindingstype gebruikt in plaats van een orchestrationTrigger bindingstype.

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

De implementatie van E1_SayHello is een relatief triviale tekenreeksopmaakbewerking.

E1_SayHello/__init__.py

def main(name: str) -> str:
    return f"Hello {name}!"

In tegenstelling tot de orchestratorfunctie heeft een activiteitsfunctie geen speciale installatie nodig. De invoer die door de orchestratorfunctie wordt doorgegeven, is rechtstreeks toegankelijk als parameter voor de functie.

PowerShell-voorbeeld komt binnenkort beschikbaar.

Java voorbeeld binnenkort beschikbaar.

Activiteiten in de Durable Task SDK nemen over van TaskActivity<TInput, TOutput>:

using Microsoft.DurableTask;
using Microsoft.Extensions.Logging;

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

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

    public override Task<string> RunAsync(TaskActivityContext context, string name)
    {
        _logger.LogInformation("Activity SayHello called with name: {Name}", name);
        return Task.FromResult($"Hello {name}!");
    }
}

[DurableTask]
public class ProcessGreetingActivity : TaskActivity<string, string>
{
    public override Task<string> RunAsync(TaskActivityContext context, string greeting)
    {
        return Task.FromResult($"{greeting} How are you today?");
    }
}

[DurableTask]
public class FinalizeResponseActivity : TaskActivity<string, string>
{
    public override Task<string> RunAsync(TaskActivityContext context, string response)
    {
        return Task.FromResult($"{response} I hope you're doing well!");
    }
}

Gebruik afhankelijkheidsinjectie om services zoals ILogger. Voeg het [DurableTask] kenmerk toe om de activiteit te registreren bij de worker.

Activiteiten in de Durable Task SDK zijn eenvoudige functies:

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

const sayHello = async (_ctx: ActivityContext, name: string): Promise<string> => {
  return `Hello ${name}!`;
};

const processGreeting = async (_ctx: ActivityContext, greeting: string): Promise<string> => {
  return `${greeting} How are you today?`;
};

const finalizeResponse = async (_ctx: ActivityContext, response: string): Promise<string> => {
  return `${response} I hope you're doing well!`;
};

In tegenstelling tot orchestrators kunnen activiteiten I/O-bewerkingen uitvoeren, zoals HTTP-aanroepen, databasequery's en bestandstoegang. De invoer wordt rechtstreeks doorgegeven als een parameter.

Activiteiten in de Durable Task SDK zijn eenvoudige functies:

from durabletask import task

def say_hello(ctx: task.ActivityContext, name: str) -> str:
    """First activity that greets the user."""
    return f"Hello {name}!"

def process_greeting(ctx: task.ActivityContext, greeting: str) -> str:
    """Second activity that processes the greeting."""
    return f"{greeting} How are you today?"

def finalize_response(ctx: task.ActivityContext, response: str) -> str:
    """Third activity that finalizes the response."""
    return f"{response} I hope you're doing well!"

In tegenstelling tot orchestrators kunnen activiteiten I/O-bewerkingen uitvoeren, zoals HTTP-aanroepen, databasequery's en bestandstoegang. De invoer wordt rechtstreeks doorgegeven als een parameter.

Dit voorbeeld wordt weergegeven voor .NET, JavaScript, Java en Python.

Activiteiten in Java worden gedefinieerd met behulp van TaskActivityFactory:

import com.microsoft.durabletask.*;

.addActivity(new TaskActivityFactory() {
    @Override
    public String getName() { return "Reverse"; }

    @Override
    public TaskActivity create() {
        return ctx -> {
            String input = ctx.getInput(String.class);
            StringBuilder builder = new StringBuilder(input);
            builder.reverse();
            return builder.toString();
        };
    }
})
.addActivity(new TaskActivityFactory() {
    @Override
    public String getName() { return "Capitalize"; }

    @Override
    public TaskActivity create() {
        return ctx -> ctx.getInput(String.class).toUpperCase();
    }
})
.addActivity(new TaskActivityFactory() {
    @Override
    public String getName() { return "ReplaceWhitespace"; }

    @Override
    public TaskActivity create() {
        return ctx -> {
            String input = ctx.getInput(String.class);
            return input.trim().replaceAll("\\s", "-");
        };
    }
})

Registreer elke activiteit met de worker builder door gebruik te maken van addActivity. Activiteiten kunnen I/O-bewerkingen uitvoeren en resultaten retourneren aan de orchestrator.

public static class HttpStart
{
    [FunctionName("HttpStart")]
    public static async Task<HttpResponseMessage> Run(
        [HttpTrigger(AuthorizationLevel.Function, methods: "post", Route = "orchestrators/{functionName}")] HttpRequestMessage req,
        [DurableClient] IDurableClient starter,
        string functionName,
        ILogger log)
    {
        // Function input comes from the request content.
        object eventData = await req.Content.ReadAsAsync<object>();
        string instanceId = await starter.StartNewAsync(functionName, eventData);

        log.LogInformation($"Started orchestration with ID = '{instanceId}'.");

        return starter.CreateCheckStatusResponse(req, instanceId);
    }
}

Als u wilt communiceren met orchestrators, voegt u een DurableClient invoerbinding toe. Gebruik de client om een orchestratie te starten en een HTTP-antwoord te retourneren dat URL's bevat waarmee u de status van de nieuwe orchestratie kunt controleren.

HttpStart/function.json

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "authLevel": "anonymous",
      "name": "req",
      "type": "httpTrigger",
      "direction": "in",
      "route": "orchestrators/{functionName}",
      "methods": [
        "post",
        "get"
      ]
    },
    {
      "name": "$return",
      "type": "http",
      "direction": "out"
    },
    {
      "name": "starter",
      "type": "durableClient",
      "direction": "in"
    }
  ]
}

Als u wilt communiceren met orchestrators, moet de functie een durableClient invoerbinding bevatten.

HttpStart/__init__.py

import logging

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


async def main(req: func.HttpRequest, starter: str) -> func.HttpResponse:
    client = df.DurableOrchestrationClient(starter)
    instance_id = await client.start_new(req.route_params["functionName"], None, None)

    logging.info(f"Started orchestration with ID = '{instance_id}'.")

    return client.create_check_status_response(req, instance_id)

Gebruik de DurableOrchestrationClient constructor om een Durable Functions-client te maken. Gebruik de client om een orchestratie te starten en een HTTP-antwoord te retourneren dat URL's bevat waarmee u de status van de nieuwe orchestratie kunt controleren.

PowerShell-voorbeeld komt binnenkort beschikbaar.

Java voorbeeld binnenkort beschikbaar.

using Microsoft.DurableTask.Client;

// Create the client
var client = DurableTaskClientBuilder.UseDurableTaskScheduler(connectionString).Build();

// Schedule a new orchestration instance
string instanceId = await client.ScheduleNewOrchestrationInstanceAsync(
    nameof(GreetingOrchestration),
    input: "World");

Console.WriteLine($"Started orchestration with ID: {instanceId}");

// Wait for the orchestration to complete
OrchestrationMetadata result = await client.WaitForInstanceCompletionAsync(
    instanceId,
    getInputsAndOutputs: true);

Console.WriteLine($"Orchestration completed with result: {result.ReadOutputAs<string>()}");

Maak de DurableTaskClient met behulp van een verbindingsreeks naar de Durable Task Scheduler. Gebruik ScheduleNewOrchestrationInstanceAsync om een orkestratie te starten en WaitForInstanceCompletionAsync om te wachten op voltooiing.

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

// Schedule a new orchestration instance
const instanceId = await client.scheduleNewOrchestration(greetingOrchestrator, "World");
console.log(`Started orchestration with ID: ${instanceId}`);

// Wait for the orchestration to complete
const state = await client.waitForOrchestrationCompletion(instanceId, true, 30);
console.log(`Orchestration completed with result: ${state?.serializedOutput}`);

Maak de DurableTaskAzureManagedClientBuilder met behulp van een verbindingsreeks naar de Durable Task Scheduler. Gebruik scheduleNewOrchestration om een orchestratie te starten en waitForOrchestrationCompletion om te wachten op voltooiing.

from durabletask.azuremanaged.client import DurableTaskSchedulerClient

# Create the client
client = DurableTaskSchedulerClient(
    host_address=endpoint,
    secure_channel=endpoint != "http://localhost:8080",
    taskhub=taskhub_name,
    token_credential=credential
)

# Schedule a new orchestration instance
instance_id = client.schedule_new_orchestration(
    function_chaining_orchestrator,
    input="World"
)

print(f"Started orchestration with ID: {instance_id}")

# Wait for the orchestration to complete
result = client.wait_for_orchestration_completion(instance_id, timeout=60)

if result and result.runtime_status == OrchestrationStatus.COMPLETED:
    print(f"Orchestration completed with result: {result.serialized_output}")

De DurableTaskSchedulerClient maakt verbinding met de Durable Task Scheduler. Gebruik schedule_new_orchestration om een orkestratie te starten en wait_for_orchestration_completion om te wachten op voltooiing.

Dit voorbeeld wordt weergegeven voor .NET, JavaScript, Java en Python.

import com.microsoft.durabletask.*;
import com.microsoft.durabletask.azuremanaged.DurableTaskSchedulerClientExtensions;

// Create the client
DurableTaskClient client = DurableTaskSchedulerClientExtensions
    .createClientBuilder(connectionString)
    .build();

// Schedule a new orchestration instance
String instanceId = client.scheduleNewOrchestrationInstance(
    "ActivityChaining",
    new NewOrchestrationInstanceOptions().setInput("Hello, world!"));

System.out.println("Started orchestration with ID: " + instanceId);

// Wait for the orchestration to complete
OrchestrationMetadata result = client.waitForInstanceCompletion(
    instanceId,
    Duration.ofSeconds(30),
    true);

System.out.println("Orchestration completed with result: " + result.readOutputAs(String.class));

Maak de DurableTaskClient met behulp van een connection string. Gebruik scheduleNewOrchestrationInstance om een orkestratie te starten en waitForInstanceCompletion te wachten op voltooiing.