Funktionskette

• Schließen Sie den Schnellstartartikel ab.
• Clonen oder laden Sie das Beispielprojekt von GitHub herunter

• Schließen Sie den Schnellstartartikel ab.
• Clonen oder laden Sie das Beispielprojekt von GitHub herunter

• Schließen Sie den Schnellstartartikel ab.
• Clonen oder laden Sie das Beispielprojekt von GitHub herunter

• Schließen Sie den Schnellstartartikel ab.

[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#-Orchestrierungsfunktionen müssen über einen Parameter vom Typ DurableOrchestrationContext verfügen, der in der Assembly Microsoft.Azure.WebJobs.Extensions.DurableTask vorhanden ist. Mit diesem Kontextobjekt können Sie andere Funktionen vom Typ Aktivität aufrufen und Eingabeparameter mit der zugehörigen CallActivityAsync-Methode übergeben.

Im Code wird E1_SayHello dreimal nacheinander mit unterschiedlichen Parameterwerten aufgerufen. Der Rückgabewert jedes Aufrufs wird der Liste outputs hinzugefügt, die am Ende der Funktion zurückgegeben wird.

function.json

Wenn Sie Visual Studio Code oder das Azure Portal für die Entwicklung verwenden, finden Sie hier den Inhalt der Datei function.json für die Orchestratorfunktion. Die meisten function.json-Dateien für Orchestratoren ähneln dieser Datei stark.

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

Ein wichtiger Punkt ist der Bindungstyp orchestrationTrigger. Alle Orchestratorfunktionen müssen diesen Triggertyp verwenden.

__init__.py

Hier ist die Orchestratorfunktion:

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 Orchestrierungsfunktionen müssen die durable-functions package enthalten. Es ist eine Bibliothek, mit der Sie Durable Functions in Python schreiben können. Zwei wichtige Unterschiede zwischen einer Orchestratorfunktion und anderen Python Funktionen:

1. Die Orchestratorfunktion ist eine Generatorfunktion.
2. Die Datei registriert die Orchestratorfunktion durch Angabe main = df.Orchestrator.create(<orchestrator function name>) am Ende der Datei. Dadurch können Sie sie von anderen Hilfsfunktionen unterscheiden, die in der Datei deklariert sind.

Mit diesem context-Objekt können Sie andere Funktionen vom Typ Aktivität aufrufen und Eingabeparameter mit der zugehörigen call_activity-Methode übergeben. Im Code wird E1_SayHello dreimal nacheinander mit unterschiedlichen Parameterwerten aufgerufen. Dabei wird yield verwendet, um anzugeben, dass für die Ausführung auf die Rückgabe der asynchronen Aufrufe der Aktivitätsfunktion gewartet werden muss. Der Rückgabewert jedes Aufrufs wird am Ende der Funktion zurückgegeben.

PowerShell-Beispiel ist noch nicht verfügbar.

Java Beispiel ist noch nicht verfügbar.

Dieser Code zeigt einen Orchestrator, der drei Aktivitäten sequenziert aufruft und jede Ausgabe an die nächste Aktivität übergibt:

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 Orchestratoren erben von TaskOrchestrator<TInput, TOutput>. TaskOrchestrationContext ermöglicht es Ihnen, Aktivitäten mit CallActivityAsync aufzurufen. Der Code ruft drei Aktivitäten in Sequenz auf, wobei jede Aktivität die Ausgabe der vorherigen Aktivität empfängt.

Der folgende Code zeigt einen Orchestrator, der drei Aktivitäten nacheinander aufruft.

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-Orchestratoren sind Generatorfunktionen (async function*), die yield zum Aufrufen von Aktivitäten verwenden. Die Methode callActivity des Orchestrierungskontexts plant die Ausführung der Aktivität. Der Code ruft drei Aktivitäten in Sequenz auf, wobei die Ausgabe jeder Aktivität an die nächste übergeben wird.

Der folgende Code zeigt einen Orchestrator, der drei Aktivitäten nacheinander aufruft.

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 Orchestratoren sind Generatorfunktionen, die yield verwenden, um Aktivitäten aufzurufen. Die Methode des Orchestrierungskontexts call_activity plant die Aktivitätsausführung. Der Code ruft drei Aktivitäten in Sequenz auf, wobei die Ausgabe jeder Aktivität an die nächste übergeben wird.

Dieses Beispiel wird für .NET, JavaScript, Java und Python gezeigt.

Der folgende Code zeigt einen Orchestrator, der drei Aktivitäten nacheinander aufruft.

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 werden Orchestratoren mithilfe von TaskOrchestrationFactory definiert. Die Methode des Kontexts plant die Ausführung der callActivity Aktivität und await() wartet auf das Ergebnis. Der Code ruft drei Aktivitäten in Sequenz auf, wobei die Ausgabe jeder Aktivität an die nächste übergeben wird.

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

Aktivitäten verwenden das ActivityTrigger-Attribut. Nutz IDurableActivityContext für Aktivitätsaktionen, wie zum Beispiel um Eingaben mit GetInput<T> zu lesen.

E1_SayHello formatiert eine Grußzeichenfolge.

Anstatt eine Bindung an IDurableActivityContext, binden Sie direkt an den Typ, der an die Aktivitätsfunktion übergeben wird. Beispiel:

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

E1_SayHello/function.json

Die Datei function.json für die Aktivitätsfunktion E1_SayHello ähnelt der Datei von E1_HelloSequence, aber es wird nicht der Bindungstyp activityTrigger, sondern orchestrationTrigger verwendet.

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

Die Implementierung von E1_SayHello ist ein relativ einfacher Zeichenfolgen-Formatierungsvorgang.

E1_SayHello/__init__.py

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

Im Gegensatz zu einer Orchestratorfunktion erfordert eine Aktivitätsfunktion keine spezielle Einrichtung. Die von der Orchestratorfunktion an sie übergebene Eingabe ist direkt als Parameter für die Funktion verfügbar.

PowerShell-Beispiel wird in Kürze verfügbar sein.

Java Beispiel in Kürze verfügbar.

Aktivitäten im Sdk für dauerhafte Aufgaben erben von 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!");
    }
}

Verwenden Sie die Abhängigkeitsinjektion, um Dienste wie diese ILogger zu erhalten. Fügen Sie das [DurableTask] Attribut hinzu, um die Aktivität beim Worker zu registrieren.

Aktivitäten im Durable Task SDK sind einfache Funktionen:

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

Im Gegensatz zu Orchestratoren können Aktivitäten E/A-Vorgänge wie HTTP-Aufrufe, Datenbankabfragen und Dateizugriff ausführen. Die Eingabe wird direkt als Parameter übergeben.

Aktivitäten im Durable Task SDK sind einfache Funktionen:

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!"

Im Gegensatz zu Orchestratoren können Aktivitäten E/A-Vorgänge wie HTTP-Aufrufe, Datenbankabfragen und Dateizugriff ausführen. Die Eingabe wird direkt als Parameter übergeben.

Dieses Beispiel wird für .NET, JavaScript, Java und Python gezeigt.

Aktivitäten in Java werden mithilfe von TaskActivityFactory definiert:

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

Registrieren Sie jede Aktivität mit dem Worker Builder mithilfe von addActivity. Aktivitäten können E/A-Vorgänge ausführen und Ergebnisse an den Orchestrator zurückgeben.

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

Um mit Orchestratoren zu interagieren, fügen Sie eine DurableClient Eingabebindung hinzu. Verwenden Sie den Client, um eine Orchestrierung zu starten und eine HTTP-Antwort zurückzugeben, die URLs enthält, um den Status der neuen Orchestrierung zu überprüfen.

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"
    }
  ]
}

Um mit Orchestratoren zu interagieren, muss die Funktion eine durableClient-Eingabebindung enthalten.

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)

Verwenden Sie den DurableOrchestrationClient-Konstruktor, um einen Durable Functions-Client zu erstellen. Verwenden Sie den Client, um eine Orchestrierung zu starten und eine HTTP-Antwort zurückzugeben, die URLs enthält, um den Status der neuen Orchestrierung zu überprüfen.

PowerShell-Beispiel wird in Kürze verfügbar sein.

Java Beispiel in Kürze verfügbar.

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

Erstellen Sie die DurableTaskClient mithilfe einer Verbindungszeichenfolge für den dauerhaften Aufgabenplaner. Verwenden Sie ScheduleNewOrchestrationInstanceAsync, um eine Orchestrierung zu starten, und WaitForInstanceCompletionAsync, um auf den Abschluss zu warten.

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

Erstellen Sie die DurableTaskAzureManagedClientBuilder mithilfe einer Verbindungszeichenfolge für den dauerhaften Aufgabenplaner. Verwenden Sie scheduleNewOrchestration zum Starten einer Orchestrierung und waitForOrchestrationCompletion zum Warten auf deren Abschluss.

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

Die DurableTaskSchedulerClient verbindet sich mit dem Durable Task Scheduler. Verwenden Sie schedule_new_orchestration, um eine Orchestrierung zu starten, und wait_for_orchestration_completion, um auf die Fertigstellung zu warten.

Dieses Beispiel wird für .NET, JavaScript, Java und Python gezeigt.

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

Erstellen Sie die DurableTaskClient mithilfe einer Verbindungszeichenfolge. Verwenden Sie scheduleNewOrchestrationInstance, um eine Orchestrierung zu starten, und waitForInstanceCompletion, um auf den Abschluss zu warten.