Ewige Orchestrierung

Ewige Orchestrierungen sind Orchestratorfunktionen, die nie enden. Sie sind nützlich, wenn Sie Durable Functions für Aggregatoren und jedes Szenario verwenden möchten, das eine Unendliche Schleife erfordert.

Ewige Orchestrierungen sind Orchestrierungen , die nie enden. Sie sind nützlich, wenn Sie langlebige Orchestrierungen für Aggregatoren und Szenarien einsetzen möchten, für die eine Endlosschleife erforderlich ist.

Wie im Thema "Orchestrierungshistorie " erläutert, verfolgt das Durable Task Framework die Geschichte der einzelnen Funktionsorchester. Diese Geschichte wächst kontinuierlich, solange die Orchestratorfunktion neue Arbeiten plant. Wenn die Orchestratorfunktion in eine Endlosschleife eintritt und fortlaufend Arbeit plant, kann der Verlauf eine kritische Größe erreichen, was zu erheblichen Leistungsproblemen führen kann. Das ewige Orchestrierungskonzept wurde entwickelt, um diese Arten von Problemen für Anwendungen zu mindern, die unendliche Schleifen benötigen.

Die SDKs für dauerhafte Aufgaben verfolgen die Geschichte der einzelnen Orchestrierungen. Diese Geschichte wächst kontinuierlich, solange die Orchestrierung neue Arbeiten plant. Wenn die Orchestrierung in eine Endlosschleife eintritt und fortlaufend Arbeit plant, kann der Verlauf eine kritische Größe erreichen, was zu erheblichen Leistungsproblemen führen kann. Das ewige Orchestrierungskonzept wurde entwickelt, um diese Arten von Problemen für Anwendungen zu mindern, die unendliche Schleifen benötigen.

Anstatt Endlosschleifen zu verwenden, setzen Orchestratorfunktionen ihren Status zurück, indem sie die Methode continue-as-new in der Orchestrierungstriggerbindung aufrufen. Diese Methode verwendet einen JSON-serialisierbaren Parameter, der zur neuen Eingabe für die nächste Orchestratorfunktionsgenerierung wird.

Beim Aufrufen continue-as-newwird die Orchestrierungsinstanz mit dem neuen Eingabewert neu gestartet. Die gleiche Instanz-ID wird beibehalten, aber der Verlauf der Orchestratorfunktion wird zurückgesetzt.

Anstatt Endlosschleifen zu verwenden, setzen Orchestrierungen ihren Zustand zurück, indem sie die continue-as-new-Methode für den Orchestrierungskontext aufrufen. Diese Methode verwendet einen JSON-serialisierbaren Parameter, der zur neuen Eingabe für die nächste Orchestrierungsgenerierung wird.

Beim Aufrufen continue-as-newwird die Orchestrierungsinstanz mit dem neuen Eingabewert neu gestartet. Es wird dieselbe Instanz-ID beibehalten, aber der Verlauf der Orchestrierung wird zurückgesetzt.

Beachten Sie diese Überlegungen bei der Verwendung der continue-as-new Methode in einer Orchestrierung:

• Wenn eine Orchestratorfunktion mithilfe der continue-as-new Methode zurückgesetzt wird, verwaltet das Durable Task Framework dieselbe Instanz-ID, erstellt aber intern eine neue Ausführungs-ID und verwendet diese. Diese Ausführungs-ID wird nicht extern verfügbar gemacht, ist aber beim Debuggen der Orchestrierungsausführung hilfreich.

• Wenn während der Ausführung eine unbehandelte Ausnahme auftritt, tritt die Orchestrierung in einen fehlerhaften Zustand ein, und die Ausführung wird beendet. In diesem Zustand kann ein Aufruf von continue-as-new aus dem finally-Block einer Try-Catch-Anweisung die Orchestrierung nicht neu starten.

Beachten Sie diese Überlegungen bei der Verwendung der continue-as-new Methode in einer Orchestrierung:

• Wenn eine Orchestrierung mit der continue-as-new-Methode zurückgesetzt wird, behalten die Durable Task-SDKs dieselbe Instanz-ID bei, erstellen aber intern eine neue Ausführungs-ID. Diese Ausführungs-ID wird nicht extern verfügbar gemacht, kann aber beim Debuggen der Orchestrierungsausführung hilfreich sein.

• Wenn während der Ausführung eine unbehandelte Ausnahme auftritt, tritt die Orchestrierung in einen fehlerhaften Zustand ein, und die Ausführung wird beendet. In diesem Zustand kann ein Aufruf von continue-as-new aus dem finally-Block einer Try-Catch-Anweisung die Orchestrierung nicht neu starten.

• Die Ergebnisse unvollständiger Aufgaben werden verworfen, wenn eine Orchestrierung continue-as-new aufruft. Wenn beispielsweise ein Timer geplant ist und dann continue-as-new aufgerufen wird, bevor der Timer ausgelöst wird, wird das Timer-Ereignis verworfen.

• Optionalerweise können Sie unverarbeitete externe Ereignisse über continue-as-new Neustarts hinweg beibehalten. In .NET und Java behält continue-as-new standardmäßig nicht verarbeitete Ereignisse bei. In Python behält continue_as_new keine Ereignisse bei, es sei denn, save_events=True. In JavaScript ist ein continueAsNew Parameter (saveEvents entweder true oder false) erforderlich, um dieses Verhalten zu steuern. In allen Fällen werden unverarbeitete Ereignisse ausgeliefert, wenn die Orchestrierung das nächste Mal waitForExternalEvent oder wait_for_external_event aufruft.

[FunctionName("Periodic_Cleanup_Loop")]
public static async Task Run(
    [OrchestrationTrigger] IDurableOrchestrationContext context)
{
    await context.CallActivityAsync("DoCleanup", null);

    // sleep for one hour between cleanups
    DateTime nextCleanup = context.CurrentUtcDateTime.AddHours(1);
    await context.CreateTimer(nextCleanup, CancellationToken.None);

    context.ContinueAsNew(null);
}

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

module.exports = df.orchestrator(function*(context) {
    yield context.df.callActivity("DoCleanup");

    // sleep for one hour between cleanups
    const nextCleanup = moment.utc(context.df.currentUtcDateTime).add(1, "h");
    yield context.df.createTimer(nextCleanup.toDate());

    yield context.df.continueAsNew(undefined);
});

import azure.functions as func
import azure.durable_functions as df
from datetime import datetime, timedelta

def orchestrator_function(context: df.DurableOrchestrationContext):
    yield context.call_activity("DoCleanup")

    # sleep for one hour between cleanups
    next_cleanup = context.current_utc_datetime + timedelta(hours = 1)
    yield context.create_timer(next_cleanup)

    context.continue_as_new(None)

main = df.Orchestrator.create(orchestrator_function)

@FunctionName("Periodic_Cleanup_Loop")
public void periodicCleanupLoop(
        @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx) {
    ctx.callActivity("DoCleanup").await();

    ctx.createTimer(Duration.ofHours(1)).await();

    ctx.continueAsNew(null);
}

Der Unterschied zwischen diesem Beispiel und einer durch Zeitgeber ausgelösten Funktion besteht darin, dass Bereinigungstriggerzeiten nicht auf einem Zeitplan basieren. Beispielsweise wird ein CRON-Zeitplan, der jede Stunde eine Funktion ausführt, um 1:00 Uhr, 2:00 Uhr, 3:00 Uhr usw. ausgeführt und könnte potenziell zu Überschneidungen führen. Wenn die Bereinigung in diesem Beispiel 30 Minuten dauert, wird sie um 1:00, 2:30, 4:00 Uhr usw. angesetzt, ohne dass es zu Überschneidungen kommen kann.

public class PeriodicCleanupLoop : TaskOrchestrator<object?, object?>
{
    public override async Task<object?> RunAsync(TaskOrchestrationContext context, object? input)
    {
        await context.CallActivityAsync("DoCleanup");

        // sleep for one hour between cleanups
        await context.CreateTimer(TimeSpan.FromHours(1), CancellationToken.None);

        context.ContinueAsNew(null);
        return null;
    }
}

from datetime import timedelta
from durabletask import task

def do_cleanup(ctx: task.ActivityContext, _) -> None:
    # Cleanup logic here
    pass

def periodic_cleanup_loop(ctx: task.OrchestrationContext, _):
    yield ctx.call_activity(do_cleanup)

    # sleep for one hour between cleanups
    yield ctx.create_timer(timedelta(hours=1))

    ctx.continue_as_new(None)

public class PeriodicCleanupLoop implements TaskOrchestration {
    @Override
    public void run(TaskOrchestrationContext ctx) {
        ctx.callActivity("DoCleanup").await();

        // sleep for one hour between cleanups
        ctx.createTimer(Duration.ofHours(1)).await();

        ctx.continueAsNew(null);
    }
}

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

const doCleanup = async (_: ActivityContext, _input: void): Promise<void> => {
    // Cleanup logic here
};

const periodicCleanupLoop: TOrchestrator = async function* (ctx: OrchestrationContext): any {
    yield ctx.callActivity(doCleanup);

    // sleep for one hour between cleanups
    yield ctx.createTimer(60 * 60);

    ctx.continueAsNew(null, false);
};

Der Unterschied zwischen diesem Beispiel und einem timer-basierten Ansatz besteht darin, dass Auslösungszeiten für die Bereinigung nicht auf einem Zeitplan basieren. Beispielsweise wird ein Zeitplan, der jede Stunde ausgeführt wird, um 1:00, 2:00, 3:00 Uhr usw. ausgeführt, was zu Überlappungsproblemen führen kann. Wenn die Bereinigung in diesem Beispiel 30 Minuten dauert, wird sie um 1:00, 2:30, 4:00 Uhr usw. angesetzt, ohne dass es zu Überschneidungen kommen kann.

Verwenden Sie die Methode start-new oder schedule-new, um eine unbegrenzte Orchestrierung zu starten, genauso wie für jede andere Orchestrierungsfunktion.

[FunctionName("Trigger_Eternal_Orchestration")]
public static async Task<HttpResponseMessage> OrchestrationTrigger(
    [HttpTrigger(AuthorizationLevel.Function, "post", Route = null)] HttpRequestMessage request,
    [DurableClient] IDurableOrchestrationClient client)
{
    string instanceId = "StaticId";

    await client.StartNewAsync("Periodic_Cleanup_Loop", instanceId); 
    return client.CreateCheckStatusResponse(request, instanceId);
}

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

module.exports = async function (context, req) {
    const client = df.getClient(context);
    const instanceId = "StaticId";

    // null is used as the input, since there is no input in "Periodic_Cleanup_Loop".
    await client.startNew("Periodic_Cleanup_Loop", instanceId, null);

    context.log(`Started orchestration with ID = '${instanceId}'.`);
    return client.createCheckStatusResponse(context.bindingData.req, instanceId);
};

async def main(req: func.HttpRequest, starter: str) -> func.HttpResponse:
    client = df.DurableOrchestrationClient(starter)
    instance_id = 'StaticId'

    await client.start_new('Periodic_Cleanup_Loop', instance_id, None)

    logging.info(f"Started orchestration with ID = '{instance_id}'.")
    return client.create_check_status_response(req, instance_id)

@FunctionName("Trigger_Eternal_Orchestration")
public HttpResponseMessage triggerEternalOrchestration(
        @HttpTrigger(name = "req") HttpRequestMessage<?> req,
        @DurableClientInput(name = "durableContext") DurableClientContext durableContext) {

    String instanceID = "StaticID";
    DurableTaskClient client = durableContext.getClient();
    client.scheduleNewOrchestrationInstance("Periodic_Cleanup_Loop", null, instanceID);
    return durableContext.createCheckStatusResponse(req, instanceID);
}

Verwenden Sie die schedule-new client-Methode, um eine unbegrenzte Orchestrierung zu starten, genau wie bei jeder anderen Orchestrierung.

string instanceId = "StaticId";
await client.ScheduleNewOrchestrationInstanceAsync(
    "PeriodicCleanupLoop",
    null,
    new StartOrchestrationOptions { InstanceId = instanceId });

instance_id = "StaticId"
client.schedule_new_orchestration(periodic_cleanup_loop, instance_id=instance_id)

String instanceId = "StaticId";
client.scheduleNewOrchestrationInstance("PeriodicCleanupLoop", null, instanceId);

const instanceId = "StaticId";
await client.scheduleNewOrchestration(periodicCleanupLoop, undefined, instanceId);

Wenn eine Orchestratorfunktion schließlich abgeschlossen werden muss, rufen Sie ContinueAsNew nicht auf und lassen Sie die Funktion einfach enden.

Wenn sich eine Orchestratorfunktion in einer Endlosschleife befindet und beendet werden muss, verwenden Sie die Beendigungs-API der Orchestrierungsclientbindung, um sie zu beenden. Weitere Informationen finden Sie unter Instanzverwaltung.

Wenn eine Orchestrierung schließlich abgeschlossen werden muss, rufen Sie nicht continue-as-new auf und lassen Sie zu, dass die Orchestrierung beendet wird.

Wenn sich eine Orchestrierung in einer Endlosschleife befindet und beendet werden muss, verwenden Sie die Beenden-API auf dem Client für dauerhafte Aufgaben, um diese zu beenden.

await client.TerminateInstanceAsync(instanceId, "Cleanup no longer needed");

client.terminate_orchestration(instance_id, output="Cleanup no longer needed")

client.terminate(instanceId, "Cleanup no longer needed");

await client.terminateOrchestration(instanceId, "Cleanup no longer needed");