Dauerhafte Orchestrierungen

Durable Functions ist eine Erweiterung von Azure Functions. Mit einer Orchestratorfunktion können Sie die Ausführung anderer dauerhafter Funktionen innerhalb einer Funktions-App orchestrieren. Orchestratorfunktionen verfügen über folgende Merkmale:

• Orchestratorfunktionen definieren Funktionsworkflows unter Verwendung von prozeduralem Code. Es werden keine deklarativen Schemas oder Designer benötigt.
• Orchestratorfunktionen können andere dauerhafte Funktionen synchron und asynchron aufrufen. Ausgaben aufgerufener Funktionen können zuverlässig in lokalen Variablen gespeichert werden.
• Orchestratorfunktionen sind dauerhaft und zuverlässig. Für den Ausführungsstatus werden automatisch Prüfpunkte erstellt, wenn die Funktion wartet oder angehalten wird. Der lokale Zustand geht nie verloren, wenn der Prozess oder virtuelle Computer neu gestartet wird.
• Orchestratorfunktionen können eine lange Ausführungsdauer haben. Die Gesamtlebensdauer einer Orchestrierungsinstanz kann Sekunden, Tage oder Monate betragen oder unbegrenzt sein.

Dieser Artikel enthält eine Übersicht über Orchestratorfunktionen und erläutert, wie sie zur Bewältigung verschiedener Herausforderungen bei der App-Entwicklung beitragen können. Sollten Sie noch nicht mit den Funktionstypen vertraut sein, die in einer Durable Functions-App zur Verfügung stehen, lesen Sie zunächst den Artikel Durable Functions-Typen und -Features (Azure Functions).

Orchestrierungsidentität

Jede Instanz einer Orchestrierung verfügt über einen Instanzbezeichner (auch Instanz-ID genannt). Standardmäßig handelt es sich bei jeder Instanz-ID um eine automatisch generierte GUID. Eine Instanz-ID kann aber auch ein beliebiger benutzergenerierter Zeichenfolgenwert sein. Jede Orchestrierungsinstanz-ID muss innerhalb eines Aufgabenhubs eindeutig sein.

Im Anschluss folgen einige Regeln für Instanz-IDs:

• Instanz-IDs müssen zwischen einem und 100 Zeichen lang sein.
• Instanz-IDs dürfen nicht mit @ beginnen.
• Instanz-IDs dürfen keines der folgenden Zeichen enthalten: /, \, # und ?.
• Instanz-IDs dürfen keine Steuerzeichen enthalten.

Die Instanz-ID einer Orchestrierung ist ein erforderlicher Parameter für die meisten Instanzverwaltungsvorgänge. Darüber hinaus ist sie wichtig für Diagnosen – etwa beim Durchsuchen der Nachverfolgungsdaten für die Orchestrierung in Application Insights zu Problembehandlungs- oder Analysezwecken. Aus diesem Grund empfiehlt es sich, generierte Instanz-IDs an einem externen Speicherort (etwa in einer Datenbank oder in Anwendungsprotokollen) zu speichern, sodass später problemlos auf sie verwiesen werden kann.

Zuverlässigkeit

Orchestratorfunktionen verwalten ihren Ausführungsstatus zuverlässig mithilfe eines als Ereignissourcing bezeichneten Entwurfsmusters. Anstatt den aktuellen Zustand einer Orchestrierung direkt zu speichern, verwendet das Durable Task Framework einen ausschließlich zum Anfügen bestimmten Speicher, um die vollständige Aktionsreihe zu erfassen, die von der Funktionsorchestrierung ausgeführt wird. Ein reiner Anfügespeicher bietet viele Vorteile im Vergleich mit dem „Abladen“ des gesamten Laufzeitstatus. Zu den Vorteilen zählen die gesteigerte Leistung, Skalierbarkeit und Reaktionsfähigkeit. Sie erhalten außerdem endgültige Konsistenz von Transaktionsdaten und vollständige Überwachungspfade sowie vollständigen Verlauf. Die Überwachungspfade unterstützen zuverlässige kompensierende Aktionen.

Durable Functions verwendet Ereignissourcing transparent. Im Hintergrund gibt der await-Operator (C#) oder yield-Operator (JavaScript/Python) in einer Orchestratorfunktion die Steuerung des Orchestratorthreads an den Durable Task Framework-Verteiler zurück. Bei Java gibt es kein spezielles Schlüsselwort für die Sprache. Stattdessen gibt der Aufruf von .await() in einem Task die Steuerung über eine benutzerdefinierte Throwable-Klasse wieder an den Verteiler zurück. Der Verteiler committet dann alle neuen Aktionen, die die Orchestratorfunktion geplant hat (z.B. Aufrufen mindestens einer untergeordneten Funktion oder Planen eines permanenten Timers) in den Speicher. 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. Auf ähnliche Weise erstellt die Commitaktion Nachrichten im Speicher, um die eigentlichen Aufgaben zu planen. An diesem Punkt kann die Orchestratorfunktion aus dem Arbeitsspeicher entladen werden. Standardmäßig verwendet Durable Functions Azure Storage als Laufzeitzustandsspeicher, andere Speicheranbieter werden jedoch ebenfalls unterstützt.

Wenn eine Orchestrierungsfunktion weitere Aufgaben ausführen muss (z.B. wird eine Antwortnachricht empfangen, oder ein permanenter Timer läuft ab), wird der Orchestrator reaktiviert und führt erneut die gesamte Funktion von Beginn an neu aus, um den lokalen Status wiederherzustellen. Wenn der Code während dieser Wiedergabe versucht, eine Funktion aufzurufen (oder eine andere asynchrone Aktion auszuführen), zieht Durable Task Framework den Ausführungsverlauf der aktuellen Orchestrierung zu Rate. Wenn festgestellt wird, dass die Aktivitätsfunktion bereits ausgeführt wurde und ein Ergebnis erbracht hat, wird dieses Funktionsergebnis wiedergegeben und der Orchestratorcode weiter ausgeführt. Die Wiedergabe wird fortgesetzt, bis der Funktionscode abgeschlossen ist oder geplante neue asynchrone Arbeit ansteht.

Orchestrierungsverlauf

Das Ereignissourcingverhalten des Durable Task Frameworks ist eng an den von Ihnen geschriebenen Orchestratorfunktionscode gekoppelt. Angenommen, Sie verfügen über eine Orchestratorfunktion zur Verkettung von Aktivitäten wie etwa die folgende Orchestratorfunktion:

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

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

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

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

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

module.exports = df.orchestrator(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"));

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

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

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

    // returns ["Hello Tokyo!", "Hello Seattle!", "Hello Cairo!"]
    return output;
});

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

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]

main = df.Orchestrator.create(orchestrator_function)

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

Jedes Mal, wenn eine Aktivitätsfunktion geplant wird, erstellt das Durable Task-Framework einen Prüfpunkt für den Ausführungszustand der Funktion in einem dauerhaften Speicher-Back-End (standardmäßig Azure Table Storage). Dieser Zustand wird als Orchestrierungsverlauf bezeichnet.

Verlaufstabelle

Im Allgemeinen führt das Durable Task Framework an jedem Prüfpunkt Folgendes durch:

1. Speichert den Ausführungsverlauf in dauerhaftem Speicher.
2. Die Nachrichten für Funktionen, die der Orchestrator aufrufen möchte, werden in die Warteschlange eingereiht.
3. Nachrichten, die für den Orchestrator selbst bestimmt sind – z. B. Nachrichten des permanenten Timers –, werden in die Warteschlange eingereiht.

Nachdem der Prüfpunktvorgang abgeschlossen ist, kann die Orchestratorfunktion aus dem Arbeitsspeicher entfernt werden, bis weitere Arbeitsschritte damit ausgeführt werden müssen.

Nach Abschluss des Vorgangs sieht der Verlauf der obigen Funktion in Azure Table Storage in etwa wie in der folgenden Tabelle aus (zur Veranschaulichung gekürzt):

Einige Hinweise zu den Spaltenwerten:

• PartitionKey: Enthält die Instanz-ID der Orchestrierung.
• EventType: Steht für den Typ des Ereignisses. Detaillierte Beschreibungen aller Verlaufsereignistypen finden Sie hier.
• Timestamp: Der UTC-Zeitstempel des Verlaufsereignisses.
• Name: Der Name der Funktion, die aufgerufen wurde.
• Eingabe: Die Eingabe der Funktion im JSON-Format.
• Result: Die Ausgabe der Funktion, also ihr Rückgabewert.

Jedes Mal, wenn die Funktion nach dem Warten auf den Abschluss eines Tasks fortgesetzt wird, führt das Durable Task-Framework den Orchestrator erneut von vorne aus. Bei jeder erneuten Ausführung wird der Ausführungsverlauf herangezogen, um zu ermitteln, ob der aktuelle asynchrone Vorgang abgeschlossen wurde. Wenn der Ausführungsverlauf zeigt, dass der Task bereits abgeschlossen ist, wird gibt das Framework die Ausgabe dieses Tasks erneut wieder und geht zum nächsten Vorgang über. Dieser Prozess wird fortgesetzt, bis der gesamte Ausführungsverlauf wiedergegeben wurde. Sobald der aktuelle Ausführungsverlauf wiedergegeben wurde, werden die lokalen Variablen wieder auf ihre vorherigen Werte zurückgesetzt.

Funktionen und Muster

In den nächsten Abschnitten werden die Features und Muster von Orchestratorfunktionen beschrieben.

Untergeordnete Orchestrierungen

Orchestratorfunktionen können nicht nur Aktivitätsfunktionen aufrufen, sondern auch andere Orchestratorfunktionen. Beispielsweise können Sie aus einer Bibliothek mit Orchestratorfunktionen eine größere Orchestrierung erstellen. Alternativ können Sie mehrere Instanzen einer Orchestratorfunktion parallel ausführen.

Weitere Informationen und Beispiele finden Sie im Artikel Untergeordnete Orchestrierungen in Durable Functions (Azure Functions).

Permanente Timer

Orchestrierungen können mithilfe permanenter Timer Verzögerungen implementieren oder die Timeoutbehandlung für asynchrone Aktionen einrichten. Verwenden Sie permanente Timer in Orchestratorfunktionen anstelle von spracheigenen „Sleep“-APIs.

Weitere Informationen und Beispiele finden Sie im Artikel Timer in Durable Functions (Azure Functions).

Externe Ereignisse

Orchestratorfunktionen können auf externe Ereignisse warten, um eine Orchestrierungsinstanz zu aktualisieren. Dieses Feature von Durable Functions ist oft bei der Verarbeitung von Benutzerinteraktionen oder anderer externer Rückrufe nützlich.

Weitere Informationen und Beispiele finden Sie im Artikel Behandeln von externen Ereignissen in Durable Functions (Azure Functions).

Fehlerbehandlung

Orchestratorfunktionen können die Fehlerbehandlungsfeatures der Programmiersprache verwenden. Vorhandene Muster wie try/catch werden in Orchestrierungscode 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 im Artikel Fehlerbehandlung in Durable Functions (Azure Functions).

Kritische Abschnitte (Durable Functions 2.x, derzeit nur .NET)

Da es sich bei Orchestrierungsinstanzen um Singlethread-Instanzen handelt, müssen Sie sich keine Gedanken über Racebedingungen innerhalb einer Orchestrierung machen. Racebedingung sind jedoch möglich, wenn Orchestrierungen mit externen Systemen interagieren. Um Racebedingungen bei der Interaktion mit externen Systemen entgegenzuwirken, können Orchestratorfunktionen in .NET mithilfe einer Methode vom Typ LockAsynckritische Abschnitte definieren.

Der folgende Beispielcode zeigt eine Orchestratorfunktion, die einen kritischen Abschnitt definiert. Der kritische Abschnitt wird mithilfe der Methode LockAsync gestartet. Diese Methode erfordert die Übergabe mindestens eines Verweises an eine dauerhafte Entität, die dauerhaft den Sperrzustand verwaltet. Der Code im kritischen Abschnitt kann jeweils nur von einer einzelnen Instanz dieser Orchestrierung ausgeführt werden.

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

LockAsync ruft die dauerhaften Sperren ab und gibt IDisposable zurück, um den kritischen Abschnitt beim Verwerfen zu beenden. Dieses Ergebnis vom Typ IDisposable kann zusammen mit einem Block vom Typ using verwendet werden, um eine syntaktische Darstellung des kritischen Abschnitts zu erhalten. Wenn eine Orchestratorfunktion in einen kritischen Abschnitt eintritt, kann dieser Codeblock nur von einer einzelnen Instanz ausgeführt werden. Alle anderen Instanzen, die versuchen, in den kritischen Abschnitt einzutreten, werden blockiert, bis die vorherige Instanz den kritischen Abschnitt verlässt.

Das Feature „kritischer Abschnitt“ ist auch hilfreich, um Änderungen an dauerhaften Entitäten zu koordinieren. Weitere Informationen zu kritischen Abschnitten finden Sie im Thema Koordination von Entitäten.

Aufrufen von HTTP-Endpunkten (Durable Functions 2.x)

Von Orchestratorfunktionen dürfen keine E/A-Vorgänge ausgeführt werden, wie unter Codeeinschränkungen für Orchestratorfunktionen beschrieben. Zur Umgehung dieses Problems wird der Code, der E/A-Vorgänge ausführen muss, in der Regel in eine Aktivitätsfunktion eingeschlossen. Orchestrierungen, die mit externen Systemen interagieren, verwenden häufig Aktivitätsfunktionen, um HTTP-Aufrufe durchzuführen und das Ergebnis an die Orchestrierung zurückzugeben.

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

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

    if ((int)response.StatusCode == 400)
    {
        // handling of error codes goes here
    }
}

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

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

    if ((int)response.StatusCode == 400)
    {
        // handling of error codes goes here
    }
}

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

module.exports = df.orchestrator(function*(context) {
    const url = context.df.getInput();
    var res = yield context.df.callHttp("GET", url);
    if (res.statusCode >= 400) {
        // handling of error codes goes here
    }
});

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) {
        // handling of error codes goes here
    }
});

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

def orchestrator_function(context: df.DurableOrchestrationContext):
    url = context.get_input()
    res = yield context.call_http('GET', url)
    if res.status_code >= 400:
        # handing of error code goes here

Neben der Unterstützung grundlegender Anforderungs-/Antwortmuster unterstützt die Methode die automatische Behandlung gängiger asynchroner HTTP 202-Abrufmuster sowie die Authentifizierung mit externen Diensten unter Verwendung verwalteter Identitäten.

Weitere Informationen und ausführliche Beispiele finden Sie im Artikel HTTP-Features.

Übergeben von mehreren Parametern

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

[FunctionName("GetCourseRecommendations")]
public static async Task<object> RunOrchestrator(
    [OrchestrationTrigger] IDurableOrchestrationContext context)
{
    string major = "ComputerScience";
    int universityYear = context.GetInput<int>();

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

[FunctionName("CourseRecommendations")]
public static async Task<object> Mapper([ActivityTrigger] IDurableActivityContext inputs)
{
    // parse input for student's major and year in university
    (string Major, int UniversityYear) studentInfo = inputs.GetInput<(string, int)>();

    // retrieve and return course recommendations by major and university year
    return new
    {
        major = studentInfo.Major,
        universityYear = studentInfo.UniversityYear,
        recommendedCourses = new []
        {
            "Introduction to .NET Programming",
            "Introduction to Linux",
            "Becoming an Entrepreneur"
        }
    };
}

[Function("GetCourseRecommendations")]
public static async Task<object> RunOrchestrator(
    [OrchestrationTrigger] TaskOrchestrationContext context, int universityYear)
{
    string major = "ComputerScience";

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

[FunctionName("CourseRecommendations")]
public static async Task<object> Mapper(
    [ActivityTrigger] (string Major, int UniversityYear) studentInfo, FunctionContext executionContext)
{
    // retrieve and return course recommendations by major and university year
    return new
    {
        major = studentInfo.Major,
        universityYear = studentInfo.UniversityYear,
        recommendedCourses = new []
        {
            "Introduction to .NET Programming",
            "Introduction to Linux",
            "Becoming an Entrepreneur"
        }
    };
}

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

module.exports = df.orchestrator(function*(context) {
    const location = {
        city: "Seattle",
        state: "WA"
    };
    const weather = yield context.df.callActivity("GetWeather", location);

    // ...
});

module.exports = async function (context, location) {
    const {city, state} = location; // destructure properties into variables

    // ...
};

const getWeatherActivityName = "getWeather";

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

    // ...
});

df.app.activity(getWeatherActivityName, async function (location) {
    const { city, state } = location; // destructure properties into variables

    // ...
});

from collections import namedtuple
import azure.functions as func
import azure.durable_functions as df

def orchestrator_function(context: df.DurableOrchestrationContext):
    Location = namedtuple('Location', ['city', 'state'])
    location = Location(city='Seattle', state= 'WA')

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

    # ...

from collections import namedtuple

Location = namedtuple('Location', ['city', 'state'])

def main(location: Location) -> str:
    city, state = location
    return f"Hello {city}, {state}!"

param($Context)

$output = @()

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

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

# ...

param($location)

"Hello $($location.City), $($location.State)!"
# ...

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

@FunctionName("GetWeather")
public String getWeather(@DurableActivityTrigger(name = "location") Location location) {
    if (location.city.equals("Seattle") && location.state.equals("WA")) {
        return "Cloudy";
    } else {
        return "Unknown";
    }
}

class Location {
    public String city;
    public String state;
}

Nächste Schritte