Dauerhafte Orchestrierung

Eine Orchestratorfunktion koordiniert die Ausführung anderer Funktionen als codebasierten Workflow. Orchestrator-Funktionen weisen die folgenden Merkmale auf:

• Sie definieren Funktionsworkflows mithilfe von prozeduralen Code. Es sind keine deklarativen Schemas oder Designer erforderlich.
• Sie können andere Funktionen synchron und asynchron aufrufen. Die Ausgabe von aufgerufenen Funktionen kann in lokalen Variablen gespeichert werden.
• Sie sind darauf ausgelegt, dauerhaft und zuverlässig zu sein. Der Ausführungsfortschritt wird automatisch als Prüfpunkt gespeichert, wenn die Funktion einen await oder yield Operator aufruft. Der lokale Zustand geht nicht verloren, wenn der Prozess wiederverwendet oder der virtuelle Computer neu gestartet wird.
• Sie können lang andauernd sein. Die gesamte Lebensdauer einer Orchestrierungsinstanz kann Sekunden, Tage oder Monate sein, oder Sie können die Instanz so konfigurieren, dass sie nie endet.

In diesem Artikel erhalten Sie eine Übersicht über Orchestratorfunktionen und wie sie Ihnen dabei helfen können, unterschiedliche Herausforderungen bei der App-Entwicklung zu lösen.

Orchestrierungsidentität

Jede Instanz einer Orchestrierung weist einen Instanzbezeichner auf, der auch als Instanz-ID bezeichnet wird. Standardmäßig ist jede Instanz-ID ein automatisch generierter, global eindeutiger Bezeichner (GUID). Instanz-IDs können jedoch auch ein beliebiger vom Benutzer generierter Zeichenfolgenwert sein. Jede Orchestrierungsinstanz-ID muss innerhalb eines Aufgabenhubs eindeutig sein.

Die folgenden Regeln gelten für Instanz-IDs:

• Sie müssen zwischen 1 und 100 Zeichen bestehen.
• Sie müssen nicht mit @ beginnen.
• Sie dürfen keine /, \, # oder ? Zeichen enthalten.
• Sie dürfen keine Steuerzeichen enthalten.

Zuverlässigkeit

Orchestrator-Funktionen verwenden das Ereignisbeschaffungsdesignmuster , um ihren Ausführungszustand zuverlässig aufrechtzuerhalten. Anstatt den aktuellen Zustand einer Orchestrierung direkt zu speichern, verwendet das Durable Task Framework einen Append-Only-Speicher, um die vollständige Serie von Aktionen aufzuzeichnen, die die Funktions-Orchestrierung ausführt. Ein Append-Only-Speicher hat viele Vorteile im Vergleich zu dem vollständigen Dump des Laufzeitstatus. Zu den Vorteilen gehören erhöhte Leistung, Skalierbarkeit und Reaktionsfähigkeit. Außerdem erhalten Sie eventuelle Konsistenz für Transaktionsdaten, vollständige Prüfprotokolle und Verlauf. Die Prüfpfade unterstützen zuverlässige Ausgleichsmaßnahmen.

Das Durable Task Framework verwendet Event Sourcing transparent. Im Hintergrund verwendet eine Orchestratorfunktion einen await-Operator in C# und einen yield-Operator in JavaScript und Python. Diese Operatoren geben die Kontrolle des Orchestrator-Threads an den Dispatcher des Durable Task Frameworks zurück. In Java gibt das Aufrufen von .await() bei einer Aufgabe die Steuerung durch eine benutzerdefinierte Instanz von Throwable an den Dispatcher zurück. Der Dispatcher speichert dann alle neuen Aktionen, die von der Orchestratorfunktion geplant werden. Beispiele für Aktionen sind der Aufruf einer oder mehrerer untergeordneter Funktionen oder das Planen eines dauerhaften Timers. Die transparente Commitaktion aktualisiert den Ausführungsverlauf der Orchestrierungsinstanz, indem alle neuen Ereignisse an den Speicher angefügt werden, ähnlich wie bei einem reinen Anfügeprotokoll. Ebenso erstellt die Commit-Aktion Nachrichten im Speicher, um die eigentliche Arbeit zu planen. An diesem Punkt kann die Orchestratorfunktion aus dem Speicher entladen werden.

Wenn eine Orchestrierungsfunktion mehr Zu erledigen hat (z. B. wird eine Antwortnachricht empfangen oder ein dauerhafter Zeitgeber abläuft), wird der Orchestrator aufwachen und die gesamte Funktion von Anfang an erneut ausführen, um den lokalen Zustand neu zu erstellen. Wenn der Code während der Wiedergabe versucht, eine Funktion aufzurufen (oder andere asynchrone Arbeiten auszuführen), konsultiert das Durable Task Framework den Ausführungsverlauf der aktuellen Orchestrierung. Wenn festgestellt wird, dass die Aktivität bereits ausgeführt wurde und ein Ergebnis erzielt hat, wird das Ergebnis dieser Funktion erneut verwendet, und der Orchestratorcode läuft weiter. Die Wiedergabe wird fortgesetzt, bis der Funktionscode beendet ist oder bis eine neue asynchrone Arbeit geplant wird.

Orchestrierungsgeschichte

Das Ereignis-Sourcing-Verhalten des Durable Task Framework ist eng mit dem Orchestrator-Funktionscode verknüpft, den Sie schreiben. Angenommen, Sie verfügen über eine Orchestrierungsfunktion für die Verkettung von Aktivitäten, wie im folgenden Beispiel gezeigt.

[Function("HelloCities")]
public static async Task<List<string>> Run(
    [OrchestrationTrigger] TaskOrchestrationContext context)
{
    var outputs = new List<string>();

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

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

using Microsoft.DurableTask;

[DurableTask]
public class HelloCities : TaskOrchestrator<object?, List<string>>
{
    public override async Task<List<string>> RunAsync(
        TaskOrchestrationContext context, object? input)
    {
        var outputs = new List<string>();

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

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

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

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

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

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

df.app.orchestration("helloSequence", function* (context) {
    const output = [];
    output.push(yield context.df.callActivity("sayHello", "Tokyo"));
    output.push(yield context.df.callActivity("sayHello", "Seattle"));
    output.push(yield context.df.callActivity("sayHello", "London"));

    // Return ["Hello Tokyo!", "Hello Seattle!", "Hello London!"].
    return output;
});

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

app = df.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)

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

param($Context)

$output = @()

$output += Invoke-DurableActivity -FunctionName 'SayHello' -Input 'Tokyo'
$output += Invoke-DurableActivity -FunctionName 'SayHello' -Input 'Seattle'
$output += Invoke-DurableActivity -FunctionName 'SayHello' -Input 'London'

$output

@FunctionName("HelloCities")
public String helloCitiesOrchestrator(
        @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx) {
    String result = "";
    result += ctx.callActivity("SayHello", "Tokyo", String.class).await() + ", ";
    result += ctx.callActivity("SayHello", "Seattle", String.class).await() + ", ";
    result += ctx.callActivity("SayHello", "London", String.class).await();
    return result;
}

using Microsoft.DurableTask;

[DurableTask]
public class HelloCities : TaskOrchestrator<object?, List<string>>
{
    public override async Task<List<string>> RunAsync(TaskOrchestrationContext context, object? input)
    {
        var outputs = new List<string>();

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

        return outputs;
    }
}

from durabletask import task

def hello_cities(ctx: task.OrchestrationContext, _):
    result1 = yield ctx.call_activity("say_hello", input="Tokyo")
    result2 = yield ctx.call_activity("say_hello", input="Seattle")
    result3 = yield ctx.call_activity("say_hello", input="London")
    return [result1, result2, result3]

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

public class HelloCities implements TaskOrchestration {
    @Override
    public void run(TaskOrchestrationContext ctx) {
        String result = "";
        result += ctx.callActivity("SayHello", "Tokyo", String.class).await() + ", ";
        result += ctx.callActivity("SayHello", "Seattle", String.class).await() + ", ";
        result += ctx.callActivity("SayHello", "London", String.class).await();
        ctx.complete(result);
    }
}

Wenn eine Aktivitätsfunktion geplant ist, speichert das Durable Task Framework den Ausführungszustand der Funktion an verschiedenen Prüfpunkten. Bei jedem Prüfpunkt speichert das Framework den Zustand in einem dauerhaften Speicher-Back-End. Dieser Zustand wird als Orchestrierungsgeschichte bezeichnet.

Jedes Mal, wenn die Funktion nach dem Warten auf den Abschluss einer Aufgabe fortgesetzt wird, führt das Durable Task Framework den Orchestrator erneut von vorne aus. Bei jeder erneuten Ausführung wird der Ausführungsverlauf konsultiert, um zu ermitteln, ob die aktuelle asynchrone Aufgabe abgeschlossen ist. Wenn der Ausführungsverlauf zeigt, dass die Aufgabe bereits abgeschlossen ist, gibt das Framework die Ausgabe dieser Aufgabe wieder und fährt mit der nächsten Aufgabe fort. Dieser Prozess wird fortgesetzt, bis der gesamte Ausführungsverlauf wiedergegeben wird. Nachdem der aktuelle Ausführungsverlauf wiedergegeben wurde, werden die lokalen Variablen in ihren vorherigen Werten wiederhergestellt.

Funktionen und Muster

In den folgenden Abschnitten werden die Features und Muster von Orchestratorfunktionen beschrieben.

Sub-Orchestrierungen

Orchestratorfunktionen können Aktivitätsfunktionen aufrufen, aber auch andere Orchestratorfunktionen. Sie können beispielsweise eine größere Orchestrierung aus einer Bibliothek von Orchestratorfunktionen erstellen. Sie können auch mehrere Instanzen einer Orchestratorfunktion parallel ausführen.

Weitere Informationen und Beispiele finden Sie in Suborchestrierungen in Durable Functions (Azure Functions).

Dauerhafte Timer

Orchestrierungen können dauerhafte Timer einplanen, um Verzögerungen zu implementieren oder um eine Zeitüberschreitung bei asynchronen Aktionen einzurichten. Verwenden Sie dauerhafte Zeitgeber in Orchestratorfunktionen anstelle von spracheigenen sleep APIs.

Weitere Informationen und Beispiele finden Sie unter Timers in Durable Functions (Azure Functions).

Externe Ereignisse

Orchestratorfunktionen können warten, bis externe Ereignisse eine Orchestrierungsinstanz aktualisieren. Dieses Durable Functions-Feature ist häufig nützlich für die Behandlung von menschlichen Interaktionen oder anderen externen Rückmeldungen.

Weitere Informationen und Beispiele finden Sie unter Umgang mit externen Ereignissen in Durable Functions (Azure Functions).

Fehlerbehandlung

Orchestratorfunktionen können die Fehlerbehandlungsfeatures der Programmiersprache verwenden. Vorhandene Muster wie try/catch im Orchestrierungscode werden unterstützt.

Orchestratorfunktionen können den Aktivitäts- oder untergeordneten Orchestratorfunktionen, die sie aufrufen, auch Wiederholungsrichtlinien hinzufügen. Sollte bei einer Aktivitäts- oder untergeordneten Orchestratorfunktion ein Ausnahmefehler auftreten, kann die angegebene Wiederholungsrichtlinie die Ausführung automatisch verzögern und eine bestimmte Anzahl von Wiederholungsversuchen durchführen.

Weitere Informationen und Beispiele finden Sie unter Handling-Fehler in Durable Functions (Azure Functions).

[FunctionName("Synchronize")]
public static async Task Synchronize(
    [OrchestrationTrigger] IDurableOrchestrationContext context)
{
    var lockId = new EntityId("LockEntity", "MyLockIdentifier");
    using (await context.LockAsync(lockId))
    {
        // Critical section. Only one orchestration can enter at a time.
    }
}

[Function("CheckSiteAvailable")]
public static async Task CheckSiteAvailable(
    [OrchestrationTrigger] TaskOrchestrationContext context)
{
    Uri url = context.GetInput<Uri>();

    // Make an HTTP GET request to the specified endpoint.
    DurableHttpResponse response = await context.CallHttpAsync(
        method: HttpMethod.Get,
        uri: url,
        content: null,
        retryOptions: null);

    if ((int)response.StatusCode == 400)
    {
        // Handle error codes.
    }
}

[FunctionName("CheckSiteAvailable")]
public static async Task CheckSiteAvailable(
    [OrchestrationTrigger] IDurableOrchestrationContext context)
{
    Uri url = context.GetInput<Uri>();

    // Make an HTTP GET request to the specified endpoint.
    DurableHttpResponse response = 
        await context.CallHttpAsync(HttpMethod.Get, url);

    if ((int)response.StatusCode == 400)
    {
        // Handle error codes.
    }
}

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

df.app.orchestration("checkSiteAvailable", function* (context) {
    const url = context.df.getInput();
    var res = yield context.df.callHttp({ method: "GET", url });
    if (res.statusCode >= 400) {
        // Handle error codes.
    }
});

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

app = df.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)

@app.orchestration_trigger(context_name="context")
def orchestrator_function(context: df.DurableOrchestrationContext):
    url = context.get_input()
    res = yield context.call_http('GET', url)
    if res['statusCode'] >= 400:
        # Handle error codes.

Mehrere Parameter

Es ist nicht möglich, mehrere Parameter direkt an eine Aktivitätsfunktion zu übergeben. Es wird empfohlen, ein Array mit Objekten oder mit zusammengesetzten Objekten einzugeben.

public record CourseInfo(string Major, int UniversityYear);

[Function("GetCourseRecommendations")]
public static async Task<object> RunOrchestrator(
    [OrchestrationTrigger] TaskOrchestrationContext context)
{
    int universityYear = context.GetInput<int>();
    CourseInfo courseInfo = new("ComputerScience", universityYear);
    object courseRecommendations = await context.CallActivityAsync<object>(
        "CourseRecommendations", courseInfo);
    return courseRecommendations;
}

public class CourseInfo
{
    public string Major { get; set; }
    public int UniversityYear { get; set; }
}

[FunctionName("GetCourseRecommendations")]
public static async Task<object> RunOrchestrator(
    [OrchestrationTrigger] IDurableOrchestrationContext context)
{
    var input = new CourseInfo
    {
        Major = "ComputerScience",
        UniversityYear = context.GetInput<int>()
    };

    object courseRecommendations = await context.CallActivityAsync<object>(
        "CourseRecommendations",
        input);
    return courseRecommendations;
}

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

df.app.orchestration("getWeatherOrchestrator", function* (context) {
    const location = {
        city: "Seattle",
        state: "WA",
    };
    const weather = yield context.df.callActivity("getWeather", location);

    // ...
});

df.app.activity("getWeather", {
    handler: async (input) => {
        const { city, state } = input; // Destructure properties into variables.
        // ...
    }
});

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

app = df.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)

@app.orchestration_trigger(context_name="context")
def orchestrator_function(context: df.DurableOrchestrationContext):
    location = {'city': 'Seattle', 'state': 'WA'}

    weather = yield context.call_activity("GetWeather", location)
    # ...

@app.activity_trigger(input_name="location")
def GetWeather(location):
    city = location['city']
    state = location['state']
    return f"Hello {city}, {state}!"

param($Context)

$location = @{
    City = 'Seattle'
    State  = 'WA'
}

Invoke-DurableActivity -FunctionName 'GetWeather' -Input $location

# ...

@FunctionName("GetWeatherOrchestrator")
public String getWeatherOrchestrator(
        @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx) {
    var location = new Location();
    location.city = "Seattle";
    location.state = "WA";
    String weather = ctx.callActivity("GetWeather", location, String.class).await();
    return weather;
}

using Microsoft.DurableTask;

public record LocationInfo(string City, string State);

[DurableTask]
public class GetWeatherOrchestration : TaskOrchestrator<object?, string>
{
    public override async Task<string> RunAsync(TaskOrchestrationContext context, object? input)
    {
        var location = new LocationInfo("Seattle", "WA");
        string weather = await context.CallActivityAsync<string>("GetWeather", location);
        return weather;
    }
}

from durabletask import task

def get_weather_orchestrator(ctx: task.OrchestrationContext, _):
    location = {"city": "Seattle", "state": "WA"}
    weather = yield ctx.call_activity("get_weather", input=location)
    # ...

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

public class GetWeatherOrchestration implements TaskOrchestration {
    @Override
    public void run(TaskOrchestrationContext ctx) {
        Location location = new Location();
        location.city = "Seattle";
        location.state = "WA";
        String weather = ctx.callActivity("GetWeather", location, String.class).await();
        ctx.complete(weather);
    }
}

Nächste Schritte