Orchestration példányok kezelése

A beépített példánykezelési API-k használatával elindíthatja, lekérdezheti, leállíthatja, felfüggesztheti, folytathatja és törölheti az orchestration példányokat a tartós munkafolyamatokban. A Durable Functions orchestration klienskötés elérhetővé teszi ezeket az API-kat. A Durable Task SDK-kban ugyanazok a műveletek érhetők el az DurableTaskClient osztályon keresztül. Ez a cikk bemutatja, hogyan hajthat végre minden egyes példánykezelési műveletet kód példákkal mindkét platformon.

Jótanács

A Azure Durable Task Scheduler a Durable Functions és a Durable Task SDK-k számára is ajánlott háttérrendszer, amely teljes mértékben felügyelt, kiszolgáló nélküli felületet biztosít a tartós munkafolyamatok nagy léptékű futtatásához.

Példányok indítása

A vezénylési ügyfél start-new (vagy schedule-new) metódusa új vezénylési példányt indít el. Ez a metódus belsőleg egy üzenetet ír a konfigurált háttérrendszerbe (például a Durable Task Scheduler vagy a Azure Storage), majd visszaadja. Ez az üzenet aszinkron módon elindít egy folyamatvezérlési eljárást a megadott névvel.

Az új orchestrációs példány indításához szükséges paraméterek a következők:

Név: Az ütemezni kívánt vezénylő függvény neve.
Bemenet: Minden olyan JSON-szerializálható adat, amelyet a vezénylő függvény bemeneteként kell átadni.
InstanceId: (Nem kötelező) A példány egyedi azonosítója. Ha nem adja meg ezt a paramétert, a metódus véletlenszerű azonosítót használ.

Jótanács

Amikor csak lehetséges, használjon véletlenszerű azonosítót a példányazonosítóhoz. A véletlenszerű példányazonosítók segítenek egyenlő terheléselosztást biztosítani a vezénylő függvények több virtuális gépen való skálázása során. A nem véletlenszerűen generált példányazonosítók használatának megfelelő ideje az, amikor az azonosító külső forrásból származik, vagy amikor a singleton vezénylőmintát implementálja.

Név: Az ütemezni kívánt vezénylés neve.
Bemenet: Minden olyan JSON-szerializálható adat, amelyet a vezénylés bemeneteként kell átadni.
InstanceId: (Nem kötelező) A példány egyedi azonosítója. Ha nem adja meg ezt a paramétert, a metódus véletlenszerű azonosítót használ.

Jótanács

Amikor csak lehetséges, használjon véletlenszerű azonosítót a példányazonosítóhoz. A véletlenszerű példányazonosítók segítenek egyenlő terheléselosztást biztosítani a vezénylések több virtuális gépen való skálázásakor. A nem véletlenszerűen generált példányazonosítók használatának megfelelő ideje az, amikor az azonosító külső forrásból származik, vagy amikor a singleton vezénylőmintát implementálja.

A következő példafüggvény elindít egy új vezénylési példányt:

[FunctionName("HelloWorldQueueTrigger")]
public static async Task Run(
    [QueueTrigger("start-queue")] string input,
    [DurableClient] IDurableOrchestrationClient starter,
    ILogger log)
{
    string instanceId = await starter.StartNewAsync("HelloWorld", input);
    log.LogInformation($"Started orchestration with ID = '{instanceId}'.");
}

Megjegyzés:

Az előző C#-kód a folyamaton belüli modellt IDurableOrchestrationClienthasználja, amely elavultként van megjelölve a Durable Functions bővítmény újabb verzióiban. Új .NET-projektek esetén fontolja meg a .NET izolált feldolgozói modellDurableTaskClienthasználatát. További információkért tekintse meg a Durable Functions-verziókról szóló cikket.

Ha másként nincs megadva, az ezen a lapon szereplő példák a HTTP-eseményindítót az alábbi function.jsonhasználják.

function.json

{
  "bindings": [
    {
      "name": "req",
      "type": "httpTrigger",
      "direction": "in",
      "methods": ["post"]
    },
    {
      "name": "$return",
      "type": "http",
      "direction": "out"
    },
    {
      "name": "starter",
      "type": "durableClient",
      "direction": "in"
    }
  ],
  "disabled": false
}

Megjegyzés:

Ez a példa a Durable Functions 2.x-es verzióját célozza meg. Használja az orchestrationClient-t a durableClient helyett az 1.x verzióban.

index.js

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

module.exports = async function(context, input) {
    const client = df.getClient(context);

    const instanceId = await client.startNew("HelloWorld", undefined, input);
    context.log(`Started orchestration with ID = ${instanceId}.`);
};

Ha másként nincs megadva, az ezen a lapon szereplő példák a HTTP-eseményindítót az alábbi function.jsonhasználják.

function.json

{
  "scriptFile": "__init__.py",
  "bindings": [    
    {
      "name": "req",
      "type": "httpTrigger",
      "direction": "in",
      "methods": ["post"]
    },
    {
      "name": "$return",
      "type": "http",
      "direction": "out"
    },
    {
      "name": "starter",
      "type": "durableClient",
      "direction": "in"
    }
  ],
  "disabled": false
}

Megjegyzés:

Ez a példa a Durable Functions 2.x-es verzióját célozza meg. Használja az orchestrationClient-t a durableClient helyett az 1.x verzióban.

__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('HelloWorld', None, None)
    logging.log(f"Started orchestration with ID = ${instance_id}.")

Ha másként nincs megadva, az ezen a lapon szereplő példák a HTTP-eseményindítót az alábbi function.jsonhasználják.

function.json

{
  "bindings": [    
    {
      "name": "Request",
      "type": "httpTrigger",
      "direction": "in",
      "methods": ["post"]
    },
    {
      "name": "Response",
      "type": "http",
      "direction": "out"
    },
    {
      "name": "starter",
      "type": "durableClient",
      "direction": "in"
    }
  ],
  "disabled": false
}

Megjegyzés:

Ez a példa a Durable Functions 2.x-es verzióját célozza meg. Használja az orchestrationClient-t a durableClient helyett az 1.x verzióban.

run.ps1

param($Request, $TriggerMetadata)

$InstanceId = Start-DurableOrchestration -FunctionName 'HelloWorld'
Write-Host "Started orchestration with ID = '$InstanceId'"

@FunctionName("HelloWorldQueueTrigger")
public void helloWorldQueueTrigger(
        @QueueTrigger(name = "input", queueName = "start-queue", connection = "Storage") String input,
        @DurableClientInput(name = "durableContext") DurableClientContext durableContext,
        final ExecutionContext context) {
    DurableTaskClient client = durableContext.getClient();        
    String instanceID = client.scheduleNewOrchestrationInstance("HelloWorld");
    context.getLogger().info("Scheduled orchestration with ID = " + instanceID);
}

Amikor meg szeretné várni, hogy az orchestration elinduljon, mielőtt visszatér a függvényből, használja az waitForInstanceStart() metódust.

// wait up to 30 seconds for the scheduled orchestration to enter the "Running" state
client.waitForInstanceStart(instanceID, Duration.ofSeconds(30));

Fontos

A PowerShell Durable Task SDK jelenleg nem érhető el.

Az alábbi kód bemutatja, hogyan indíthat el egy új vezénylési példányt a Durable Task SDK-k használatával:

using Microsoft.DurableTask.Client;

// Schedule a new orchestration instance
string instanceId = await client.ScheduleNewOrchestrationInstanceAsync("HelloWorld", input);
Console.WriteLine($"Started orchestration with ID = '{instanceId}'.");

// Optionally, wait for the orchestration to start
OrchestrationMetadata metadata = await client.WaitForInstanceStartAsync(instanceId, timeout: TimeSpan.FromSeconds(30));

from durabletask.azuremanaged.client import DurableTaskSchedulerClient

# Schedule a new orchestration instance
instance_id = client.schedule_new_orchestration(hello_world, input=input_data)
print(f"Started orchestration with ID = '{instance_id}'.")

# Optionally, wait for the orchestration to start
state = client.wait_for_orchestration_start(instance_id, timeout=30)

import com.microsoft.durabletask.DurableTaskClient;

// Schedule a new orchestration instance
String instanceId = client.scheduleNewOrchestrationInstance("HelloWorld", input);
System.out.println("Started orchestration with ID = '" + instanceId + "'.");

// Optionally, wait for the orchestration to start
OrchestrationMetadata metadata = client.waitForInstanceStart(instanceId, Duration.ofSeconds(30), false);

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

const client = createAzureManagedClient(connectionString);

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

// Optionally, wait for the orchestration to start
const state = await client.waitForOrchestrationStart(instanceId, false, 30);

Lekérdezéspéldányok

Az új vezénylési példányok elindítása után valószínűleg le kell kérdeznie a futtatókörnyezet állapotát, hogy megtudja, futnak-e, befejezték-e vagy sikertelenek.

A vezénylési ügyfél get-status metódusa egy vezénylési példány állapotát adja vissza.

Paraméterként egy instanceId (kötelező), showHistory (nem kötelező), showHistoryOutput (nem kötelező) és showInput (nem kötelező) paramétert használ.

showHistory: Ha be van trueállítva, a válasz tartalmazza a végrehajtási előzményeket.
showHistoryOutput: Ha be van trueállítva, a végrehajtási előzmények tevékenységkimeneteket tartalmaznak.
showInput: Ha be van falseállítva, a válasz nem tartalmazza a függvény bemenetét. Az alapértelmezett érték a true.

A metódus a következő tulajdonságokkal rendelkező objektumot ad vissza:

Név: A vezénylő függvény neve.
InstanceId: A vezénylési példányazonosító (meg kell egyeznie a instanceId bemenet értékével).
CreatedTime: Az időpont, amikor az orchestrator függvény elindul.
LastUpdatedTime: Az időpont, amikor az orchestration utolsó ellenőrzőpontja történt.
Bemenet: A függvény bemenete JSON-értékként. Ez a mező nincs kitöltve, ha showInputfalse.
CustomStatus: Egyéni orkesztáció állapota JSON formátumban.
Kimenet: A függvény kimenete JSON-értékként (ha a függvény befejeződik). Ha a vezénylő függvény meghibásodik, ez a tulajdonság tartalmazza a hiba részleteit. Ha a vezénylő függvényt felfüggesztik vagy megszakítják, ez a tulajdonság tartalmazza a felfüggesztés vagy a megszüntetés okát (ha van ilyen).
RuntimeStatus: Az alábbi értékek egyike:
- Függőben: A példány ütemezett, de még nem indult el.
- Fut: A példány fut.
- Befejezve: A folyamat normálisan befejeződött.
- ContinuedAsNew: A példány új történelemmel indult újra. Ez az állapot átmeneti állapot.
- Sikertelen: A példány hiba miatt meghiúsult.
- Leállt: A példány hirtelen megállt.
- Felfüggesztve: A példány jelenleg fel van függesztve, és később újra folytatható.
Előzmények: A vezénylés végrehajtási története. Ezt a mezőt csak akkor tölti ki a rendszer, ha showHistory be truevan állítva.

showHistory: Ha be van trueállítva, a válasz tartalmazza a végrehajtási előzményeket.
showHistoryOutput: Ha be van trueállítva, a végrehajtási előzmények tevékenységkimeneteket tartalmaznak.
showInput: Ha false be van állítva, a válasz nem fogja tartalmazni a vezénylés bemenetét. Az alapértelmezett érték a true.

A metódus a következő tulajdonságokkal rendelkező objektumot ad vissza:

Név: Az orchestration neve.
InstanceId: A vezénylési példányazonosító (meg kell egyeznie a instanceId bemenet értékével).
CreatedTime: Az az időpont, amikor a vezénylés elindul.
LastUpdatedTime: Az időpont, amikor az orchestration utolsó ellenőrzőpontja történt.
Bemenet: Az orkestráció bemenete JSON-értékként. Ez a mező nincs kitöltve, ha showInputfalse.
CustomStatus: Egyéni orkesztáció állapota JSON formátumban.
Kimenet: A vezénylés kimenete JSON-értékként (ha a vezénylés befejeződött). Ha a vezénylés sikertelen, ez a tulajdonság tartalmazza a hiba részleteit. Ha a vezénylés felfüggesztésre vagy megszakításra kerül, ez a tulajdonság tartalmazza a felfüggesztés vagy a megszüntetés okát (ha van ilyen).
RuntimeStatus: Az alábbi értékek egyike:
- Függőben: A példány ütemezett, de még nem indult el.
- Fut: A példány fut.
- Befejezve: A folyamat normálisan befejeződött.
- ContinuedAsNew: A példány új történelemmel indult újra. Ez az állapot átmeneti állapot.
- Sikertelen: A példány hiba miatt meghiúsult.
- Leállt: A példány hirtelen megállt.
- Felfüggesztve: A példány jelenleg fel van függesztve, és később újra folytatható.
Előzmények: A vezénylés végrehajtási története. Ezt a mezőt csak akkor tölti ki a rendszer, ha showHistory be truevan állítva.

Megjegyzés:

A vezénylő nincs megjelölve Completed, amíg az összes ütemezett feladat be nem fejeződik és a vezénylő visszatér. Más szóval, nem elegendő, ha egy orchesztrátor eléri a return állapotát ahhoz, hogy Completed-ként legyen jelölve. Ez különösen fontos a használatban lévő esetekben WhenAny ; ezek a vezénylők gyakran return az összes ütemezett feladat végrehajtása előtt.

Ez a metódus null (.NET és Java), undefined (JavaScript) vagy None (Python) értéket ad vissza, ha a példány nem létezik.

[FunctionName("GetStatus")]
public static async Task Run(
    [DurableClient] IDurableOrchestrationClient client,
    [QueueTrigger("check-status-queue")] string instanceId)
{
    DurableOrchestrationStatus status = await client.GetStatusAsync(instanceId);
    // do something based on the current status.
}

Megjegyzés:

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

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

    const status = await client.getStatus(instanceId);
    // do something based on the current status.
    // example: if status.runtimeStatus === df.OrchestrationRuntimeStatus.Running: ...
}

A function.json konfigurációjáról a Példányok indítása című témakörben olvashat.

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

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

    status = await client.get_status(instance_id)
    # do something based on the current status
    # example: if (existing_instance.runtime_status is df.OrchestrationRuntimeStatus.Running) { ...

param($Request, $TriggerMetadata)

# Get instanceid from body
$InstanceId = $Request.Body.InstanceId

$Status = Get-DurableStatus -InstanceId $InstanceId -ShowHistory -ShowHistoryOutput -ShowInput
Write-Host "Status: $($Status | ConvertTo-Json)"

# Do something based on status
# example: if ($Status.runtimeStatus -eq 'Running') { ... }

@FunctionName("GetStatus")
public void getStatus(
        @QueueTrigger(name = "instanceID", queueName = "check-status-queue", connection = "Storage") String instanceID,
        @DurableClientInput(name = "durableContext") DurableClientContext durableContext,
        final ExecutionContext context) {
    DurableTaskClient client = durableContext.getClient();        
    OrchestrationMetadata metadata = client.getInstanceMetadata(instanceID, false);
    if (metadata != null) {
        OrchestrationRuntimeStatus status = metadata.getRuntimeStatus();
        switch (status) {
            // do something based on the current status
        }
    }
}

using Microsoft.DurableTask.Client;

// Get the status of an orchestration instance
OrchestrationMetadata? metadata = await client.GetInstanceAsync(instanceId, getInputsAndOutputs: true);
if (metadata != null)
{
    OrchestrationRuntimeStatus status = metadata.RuntimeStatus;
    // do something based on the current status
}

from durabletask.azuremanaged.client import DurableTaskSchedulerClient

# Get the status of an orchestration instance
state = client.get_orchestration_state(instance_id, fetch_payloads=True)
if state is not None:
    status = state.runtime_status
    # do something based on the current status

import com.microsoft.durabletask.DurableTaskClient;
import com.microsoft.durabletask.OrchestrationMetadata;

// Get the status of an orchestration instance
OrchestrationMetadata metadata = client.getInstanceMetadata(instanceId, true);
if (metadata != null) {
    OrchestrationRuntimeStatus status = metadata.getRuntimeStatus();
    // do something based on the current status
}

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

const client = createAzureManagedClient(connectionString);

// Get the status of an orchestration instance
const state = await client.getOrchestrationState(instanceId, true);
if (state) {
    const status = state.runtimeStatus;
    // do something based on the current status
}

Az összes orchestration példány lekérdezése

A nyelvi SDK API-kkal lekérdezheti a feladatközpont összes vezénylési példányának állapotát. Ez a "list-instances" vagy "get-status" API a lekérdezési paramétereknek megfelelő orchestration példányokat képviselő objektumok listáját adja vissza.

[FunctionName("GetAllStatus")]
public static async Task Run(
    [HttpTrigger(AuthorizationLevel.Anonymous, "get", "post")] HttpRequestMessage req,
    [DurableClient] IDurableOrchestrationClient client,
    ILogger log)
{
    var noFilter = new OrchestrationStatusQueryCondition();
    OrchestrationStatusQueryResult result = await client.ListInstancesAsync(
        noFilter,
        CancellationToken.None);
    foreach (DurableOrchestrationStatus instance in result.DurableOrchestrationState)
    {
        log.LogInformation(JsonConvert.SerializeObject(instance));
    }

    // Note: ListInstancesAsync only returns the first page of results.
    // To request additional pages provide the result.ContinuationToken
    // to the OrchestrationStatusQueryCondition's ContinuationToken property.
}

Megjegyzés:

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

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

    const instances = await client.getStatusAll();
    instances.forEach((instance) => {
        context.log(JSON.stringify(instance));
    });
};

A function.json konfigurációjáról a Példányok indítása című témakörben olvashat.

import logging
import json
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)

    instances = await client.get_status_all()

    for instance in instances:
        logging.log(json.dumps(instance))

A function.json konfigurációjáról a Példányok indítása című témakörben olvashat.

@FunctionName("GetAllStatus")
public String getAllStatus(
        @HttpTrigger(name = "req", methods = {HttpMethod.GET}) HttpRequestMessage<?> req,
        @DurableClientInput(name = "durableContext") DurableClientContext durableContext) {
    DurableTaskClient client = durableContext.getClient();
    OrchestrationStatusQuery noFilter = new OrchestrationStatusQuery();
    OrchestrationStatusQueryResult result = client.queryInstances(noFilter);
    return "Found " + result.getOrchestrationState().size() + " orchestrations.";
}

using Microsoft.DurableTask.Client;

// Query all orchestration instances
AsyncPageable<OrchestrationMetadata> instances = client.GetAllInstancesAsync(new OrchestrationQuery());

await foreach (OrchestrationMetadata instance in instances)
{
    Console.WriteLine(instance.InstanceId);
}

from durabletask.azuremanaged.client import DurableTaskSchedulerClient

# Query all orchestration instances
instances = client.list_orchestrations()

for instance in instances:
    print(instance.instance_id)

import com.microsoft.durabletask.DurableTaskClient;
import com.microsoft.durabletask.OrchestrationStatusQuery;
import com.microsoft.durabletask.OrchestrationStatusQueryResult;

// Query all orchestration instances
OrchestrationStatusQuery query = new OrchestrationStatusQuery();
OrchestrationStatusQueryResult result = client.queryInstances(query);

for (OrchestrationMetadata instance : result.getOrchestrationState()) {
    System.out.println(instance.getInstanceId());
}

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

const client = createAzureManagedClient(connectionString);

// Query all orchestration instances
const instances = client.getAllInstances();

for await (const instance of instances) {
    console.log(instance.instanceId);
}

Orchestration példányok lekérdezése szűrőkkel

Mi a teendő, ha nincs szüksége minden olyan információra, amelyet egy standard példány lekérdezése biztosít? Például, mi van akkor, ha csak az orchestration létrehozási ideje vagy az orchestration futásidejű állapota érdekli? A lekérdezés szűkítése szűrők alkalmazásával.

[FunctionName("QueryStatus")]
public static async Task Run(
    [HttpTrigger(AuthorizationLevel.Anonymous, "get", "post")] HttpRequestMessage req,
    [DurableClient] IDurableOrchestrationClient client,
    ILogger log)
{
    // Get the first 100 running or pending instances that were created between 7 and 1 days ago
    var queryFilter = new OrchestrationStatusQueryCondition
    {
        RuntimeStatus = new[]
        {
            OrchestrationRuntimeStatus.Pending,
            OrchestrationRuntimeStatus.Running,
        },
        CreatedTimeFrom = DateTime.UtcNow.Subtract(TimeSpan.FromDays(7)),
        CreatedTimeTo = DateTime.UtcNow.Subtract(TimeSpan.FromDays(1)),
        PageSize = 100,
    };

    OrchestrationStatusQueryResult result = await client.ListInstancesAsync(
        queryFilter,
        CancellationToken.None);
    foreach (DurableOrchestrationStatus instance in result.DurableOrchestrationState)
    {
        log.LogInformation(JsonConvert.SerializeObject(instance));
    }
}

Megjegyzés:

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

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

    const runtimeStatus = [
        df.OrchestrationRuntimeStatus.Completed,
        df.OrchestrationRuntimeStatus.Running,
    ];
    const instances = await client.getStatusBy(
        new Date(2021, 3, 10, 10, 1, 0),
        new Date(2021, 3, 10, 10, 23, 59),
        runtimeStatus
    );
    instances.forEach((instance) => {
        context.log(JSON.stringify(instance));
    });
};

A function.json konfigurációjának kezdőpéldányait lásd.

import logging
from datetime import datetime
import json
import azure.functions as func
import azure.durable_functions as df
from azure.durable_functions.models.OrchestrationRuntimeStatus import OrchestrationRuntimeStatus

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

    runtime_status = [OrchestrationRuntimeStatus.Completed, OrchestrationRuntimeStatus.Running]

    instances = await client.get_status_by(
        datetime(2021, 3, 10, 10, 1, 0),
        datetime(2021, 3, 10, 10, 23, 59),
        runtime_status
    )

    for instance in instances:
        logging.log(json.dumps(instance))

@FunctionName("GetRunningInstances")
public String getRunningInstances(
        @HttpTrigger(name = "req", methods = {HttpMethod.GET}) HttpRequestMessage<?> req,
        @DurableClientInput(name = "durableContext") DurableClientContext durableContext) {
    DurableTaskClient client = durableContext.getClient();
    OrchestrationStatusQuery filter = new OrchestrationStatusQuery()
        .setRuntimeStatusList(List.of(OrchestrationRuntimeStatus.PENDING, OrchestrationRuntimeStatus.RUNNING))
        .setCreatedTimeFrom(Instant.now().minus(Duration.ofDays(7)))
        .setCreatedTimeTo(Instant.now().minus(Duration.ofDays(1)));
    OrchestrationStatusQueryResult result = client.queryInstances(filter);
    return "Found " + result.getOrchestrationState().size() + " orchestrations.";
}

using Microsoft.DurableTask.Client;

// Get running or pending instances created in the last 7 days
var query = new OrchestrationQuery
{
    Statuses = new[] { OrchestrationRuntimeStatus.Running, OrchestrationRuntimeStatus.Pending },
    CreatedFrom = DateTime.UtcNow.AddDays(-7),
    CreatedTo = DateTime.UtcNow.AddDays(-1),
    PageSize = 100
};

AsyncPageable<OrchestrationMetadata> instances = client.GetAllInstancesAsync(query);

await foreach (OrchestrationMetadata instance in instances)
{
    Console.WriteLine($"{instance.InstanceId}: {instance.RuntimeStatus}");
}

from durabletask.azuremanaged.client import DurableTaskSchedulerClient
from datetime import datetime, timedelta

# Get running or pending instances created in the last 7 days
instances = client.list_orchestrations(
    created_time_from=datetime.utcnow() - timedelta(days=7),
    created_time_to=datetime.utcnow() - timedelta(days=1),
    runtime_status=['RUNNING', 'PENDING']
)

for instance in instances:
    print(f"{instance.instance_id}: {instance.runtime_status}")

import com.microsoft.durabletask.DurableTaskClient;
import com.microsoft.durabletask.OrchestrationStatusQuery;
import com.microsoft.durabletask.OrchestrationRuntimeStatus;
import java.time.Duration;
import java.time.Instant;
import java.util.List;

// Get running or pending instances created in the last 7 days
OrchestrationStatusQuery filter = new OrchestrationStatusQuery()
    .setRuntimeStatusList(List.of(OrchestrationRuntimeStatus.PENDING, OrchestrationRuntimeStatus.RUNNING))
    .setCreatedTimeFrom(Instant.now().minus(Duration.ofDays(7)))
    .setCreatedTimeTo(Instant.now().minus(Duration.ofDays(1)));

OrchestrationStatusQueryResult result = client.queryInstances(filter);

for (OrchestrationMetadata instance : result.getOrchestrationState()) {
    System.out.println(instance.getInstanceId() + ": " + instance.getRuntimeStatus());
}

import { createAzureManagedClient } from "@microsoft/durabletask-js-azuremanaged";
import { OrchestrationStatus } from "@microsoft/durabletask-js";

const client = createAzureManagedClient(connectionString);

// Get running or pending instances created in the last 7 days
const sevenDaysAgo = new Date(Date.now() - 7 * 24 * 60 * 60 * 1000);
const oneDayAgo = new Date(Date.now() - 24 * 60 * 60 * 1000);

const instances = client.getAllInstances({
    statuses: [OrchestrationStatus.RUNNING, OrchestrationStatus.PENDING],
    createdFrom: sevenDaysAgo,
    createdTo: oneDayAgo,
});

for await (const instance of instances) {
    console.log(`${instance.instanceId}: ${instance.runtimeStatus}`);
}

Orchesztrációs példányok megszüntetése

Ha olyan vezénylési példánya van, amely túl sokáig tart a futtatáshoz, vagy le kell állítania, mielőtt bármilyen okból befejeződik, befejezheti.

A megszakítási API két paramétere egy példányazonosító és egy oksztring , amely naplókba és a példány állapotára ír.

[FunctionName("TerminateInstance")]
public static Task Run(
    [DurableClient] IDurableOrchestrationClient client,
    [QueueTrigger("terminate-queue")] string instanceId)
{
    string reason = "Found a bug";
    return client.TerminateAsync(instanceId, reason);
}

Megjegyzés:

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

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

    const reason = "Found a bug";
    return client.terminate(instanceId, reason);
};

A function.json konfigurációjának kezdőpéldányait lásd.

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

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

    reason = "Found a bug"
    return await client.terminate(instance_id, reason)

param($Request, $TriggerMetadata)

# Get instance id from body
$InstanceId = $Request.Body.InstanceId
$Reason = 'Found a bug'

Stop-DurableOrchestration -InstanceId $InstanceId -Reason $Reason

@FunctionName("TerminateInstance")
public void terminateInstance(
        @HttpTrigger(name = "req", methods = {HttpMethod.POST}) HttpRequestMessage<String> req,
        @DurableClientInput(name = "durableContext") DurableClientContext durableContext) {
    String instanceID = req.getBody();
    String reason = "Found a bug";
    durableContext.getClient().terminate(instanceID, reason);
}

using Microsoft.DurableTask.Client;

string reason = "Found a bug";
await client.TerminateInstanceAsync(instanceId, reason);

from durabletask.azuremanaged.client import DurableTaskSchedulerClient

reason = "Found a bug"
client.terminate_orchestration(instance_id, reason=reason)

import com.microsoft.durabletask.DurableTaskClient;

String reason = "Found a bug";
client.terminate(instanceId, reason);

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

const client = createAzureManagedClient(connectionString);

const reason = "Found a bug";
await client.terminateOrchestration(instanceId, reason);

A leállított példány végül áttér egy adott állapotba Terminated. De ez a változás nem történik meg azonnal. A leállítási művelet inkább a feladatközpontban lesz várólistára állítva, valamint az adott példány egyéb műveleteivel együtt. A példány lekérdezési API-kkal megtudhatja, hogy egy leállított példány mikor érte el ténylegesen az állapotot Terminated .

Megjegyzés:

A példánymegszüntetés jelenleg nem terjed tovább. A tevékenységfüggvények és az alvezénylések befejeződnek, függetlenül attól, hogy befejezi-e az őket nevező vezénylési példányt.

Orkesztációs példányok felfüggesztése és újraindítása

A vezénylés felfüggesztése lehetővé teszi a futó vezénylés leállítását. Az orkesztráció befejezésével ellentétben utólag folytathatja a felfüggesztett orkesztrátort.

A felfüggesztési API két paramétere egy példányazonosító és egy indoklás szöveg, amelyeket a rendszer naplókba és a példány státuszába rögzít.

[FunctionName("SuspendResumeInstance")]
public static async Task Run(
    [DurableClient] IDurableOrchestrationClient client,
    [QueueTrigger("suspend-resume-queue")] string instanceId)
{
    // To suspend an orchestration
    string suspendReason = "Need to pause workflow";
    await client.SuspendAsync(instanceId, suspendReason);

    // To resume an orchestration
    string resumeReason = "Continue workflow";
    await client.ResumeAsync(instanceId, resumeReason);
}

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

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

    // To suspend an orchestration
    const suspendReason = "Need to pause workflow";
    await client.suspend(instanceId, suspendReason);

    // To resume an orchestration
    const resumeReason = "Continue workflow";
    await client.resume(instanceId, resumeReason);
};

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

async def main(req: func.HttpRequest, starter: str, instance_id: str):
    client = df.DurableOrchestrationClient(starter)

    # To suspend an orchestration
    suspend_reason = "Need to pause workflow"
    await client.suspend(instance_id, suspend_reason)

    # To resume an orchestration
    resume_reason = "Continue workflow"
    await client.resume(instance_id, resume_reason)

param($Request, $TriggerMetadata)

$InstanceId = $Request.Body.InstanceId

# To suspend an orchestration
$SuspendReason = 'Need to pause workflow'
Suspend-DurableOrchestration -InstanceId $InstanceId -Reason $SuspendReason

# To resume an orchestration
$ResumeReason = 'Continue workflow'
Resume-DurableOrchestration -InstanceId $InstanceId -Reason $ResumeReason

Megjegyzés:

Ez a funkció csak a különálló Durable Functions PowerShell SDK-ban érhető el. A migrálási útmutatóval együtt megtekintheti az önálló SDK és az örökölt beépített SDK közötti különbséget.

@FunctionName("SuspendResumeInstance")
public void suspendResumeInstance(
        @HttpTrigger(name = "req", methods = {HttpMethod.POST}) HttpRequestMessage<String> req,
        @DurableClientInput(name = "durableContext") DurableClientContext durableContext) {
    String instanceID = req.getBody();
    DurableTaskClient client = durableContext.getClient();  

    // To suspend an orchestration
    String suspendReason = "Need to pause workflow";
    client.suspendInstance(instanceID, suspendReason);

    // To resume an orchestration
    String resumeReason = "Continue workflow";
    client.resumeInstance(instanceID, resumeReason);
}

using Microsoft.DurableTask.Client;

// To suspend an orchestration
string suspendReason = "Need to pause workflow";
await client.SuspendInstanceAsync(instanceId, suspendReason);

// To resume an orchestration
string resumeReason = "Continue workflow";
await client.ResumeInstanceAsync(instanceId, resumeReason);

from durabletask.azuremanaged.client import DurableTaskSchedulerClient

# To suspend an orchestration
suspend_reason = "Need to pause workflow"
client.suspend_orchestration(instance_id, reason=suspend_reason)

# To resume an orchestration
resume_reason = "Continue workflow"
client.resume_orchestration(instance_id, reason=resume_reason)

import com.microsoft.durabletask.DurableTaskClient;

// To suspend an orchestration
String suspendReason = "Need to pause workflow";
client.suspendInstance(instanceId, suspendReason);

// To resume an orchestration
String resumeReason = "Continue workflow";
client.resumeInstance(instanceId, resumeReason);

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

const client = createAzureManagedClient(connectionString);

// To suspend an orchestration
await client.suspendOrchestration(instanceId);

// To resume an orchestration
await client.resumeOrchestration(instanceId);

A felfüggesztett példány végül átáll a Suspended állapotba. Azonban ez az átmenet nem történik meg azonnal. Ehelyett a felfüggesztési művelet várólistára kerül a feladatközpontban az adott példány egyéb műveleteivel együtt. A példány lekérdezési API-kkal megtudhatja, hogy egy futó példány ténylegesen elérte-e az állapotot Suspended .

Amikor egy felfüggesztett vezénylő folytatódik, az állapota visszaáll Running értékre.

Események küldése példányokra

Bizonyos esetekben a vezénylő függvényeknek várniuk és figyelniük kell a külső eseményeket. Ilyenek például a monitorozási és az emberi interakciós forgatókönyvek.

Bizonyos esetekben az orchestrationoknak várniuk kell, és figyelniük kell a külső eseményeket. Ilyenek például a monitorozási és az emberi interakciós forgatókönyvek.

Eseményértesítéseket küldhet a futó példányoknak az orchestration kliens esemény indítása API-jának használatával. Az orchestrációk a külső esemény várakoztatása API használatával meghallgathatják és megválaszolhatják ezeket az eseményeket.

A emelésesemény paraméterei a következők:

Példányazonosító: A példány egyedi azonosítója.
Esemény neve: A küldendő esemény neve.
Eseményadatok: JSON-szerializálható adatcsomag, amely a példánynak küldhető.

[FunctionName("RaiseEvent")]
public static Task Run(
    [DurableClient] IDurableOrchestrationClient client,
    [QueueTrigger("event-queue")] string instanceId)
{
    int[] eventData = new int[] { 1, 2, 3 };
    return client.RaiseEventAsync(instanceId, "MyEvent", eventData);
}

Megjegyzés:

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

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

    const eventData = [ 1, 2, 3 ];
    return client.raiseEvent(instanceId, "MyEvent", eventData);
};

A function.json konfigurációjának kezdőpéldányait lásd.

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

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

    event_data = [1, 2 ,3]
    return await client.raise_event(instance_id, 'MyEvent', event_data)

param($Request, $TriggerMetadata)

# Get instance id from body
$InstanceId = $Request.Body.InstanceId
$EventName = 'MyEvent'
$EventData = @(1,2,3)

Send-DurableExternalEvent -InstanceId $InstanceId -EventName $EventName -EventData $EventData

@FunctionName("RaiseEvent")
public void raiseEvent(
        @HttpTrigger(name = "req", methods = {HttpMethod.POST}) HttpRequestMessage<String> req,
        @DurableClientInput(name = "durableContext") DurableClientContext durableContext) {
    String instanceID = req.getBody();
    String eventName = "MyEvent";
    int[] eventData = { 1, 2, 3 };
    durableContext.getClient().raiseEvent(instanceID, eventName, eventData);
}

using Microsoft.DurableTask.Client;

int[] eventData = new int[] { 1, 2, 3 };
await client.RaiseEventAsync(instanceId, "MyEvent", eventData);

from durabletask.azuremanaged.client import DurableTaskSchedulerClient

event_data = [1, 2, 3]
client.raise_orchestration_event(instance_id, "MyEvent", event_data)

import com.microsoft.durabletask.DurableTaskClient;

int[] eventData = { 1, 2, 3 };
client.raiseEvent(instanceId, "MyEvent", eventData);

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

const client = createAzureManagedClient(connectionString);

const eventData = [1, 2, 3];
await client.raiseOrchestrationEvent(instanceId, "MyEvent", eventData);

Megjegyzés:

Ha nincs összhangolási példány a megadott példányazonosítóval, az eseményüzenet elvetésre kerül. Ha egy példány létezik, de még nem várja meg az eseményt, az esemény a példány állapotában lesz tárolva, amíg készen nem áll a fogadásra és a feldolgozásra.

Várakozás a vezénylés befejezésére

Hosszú ideig futó vezénylés esetén érdemes megvárni és lekérni a vezénylés eredményét. Ezekben az esetekben hasznos egy időtúllépési korlátot meghatározni az orchestráción. Ha lejár az időkorlát, az eredmények helyett az orchestráció állapotát adjuk vissza.

A "wait for completion or create check status response" API-val szinkron módon lekérheti egy orchestration instance tényleges kimenetét. Ez a módszer alapértelmezés szerint tíz másodperces időtúllépéssel és egy másodperces lekérdezési időközzel rendelkezik.

Íme egy példa HTTP-trigger függvényre, amely bemutatja, hogyan használhatja ezt az API-t:

// Copyright (c) .NET Foundation. All rights reserved.
// Licensed under the MIT License. See LICENSE in the project root for license information.

using System;
using System.Net.Http;
using System.Threading.Tasks;
using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Extensions.DurableTask;
using Microsoft.Azure.WebJobs.Extensions.Http;
using Microsoft.Extensions.Logging;

namespace VSSample
{
    public static class HttpSyncStart
    {
        private const string Timeout = "timeout";
        private const string RetryInterval = "retryInterval";

        [FunctionName("HttpSyncStart")]
        public static async Task<HttpResponseMessage> Run(
            [HttpTrigger(AuthorizationLevel.Function, methods: "post", Route = "orchestrators/{functionName}/wait")]
            HttpRequestMessage req,
            [DurableClient] IDurableOrchestrationClient 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}'.");

            TimeSpan timeout = GetTimeSpan(req, Timeout) ?? TimeSpan.FromSeconds(30);
            TimeSpan retryInterval = GetTimeSpan(req, RetryInterval) ?? TimeSpan.FromSeconds(1);
            
            return await starter.WaitForCompletionOrCreateCheckStatusResponseAsync(
                req,
                instanceId,
                timeout,
                retryInterval);
        }

        private static TimeSpan? GetTimeSpan(HttpRequestMessage request, string queryParameterName)
        {
            string queryParameterStringValue = request.RequestUri.ParseQueryString()[queryParameterName];
            if (string.IsNullOrEmpty(queryParameterStringValue))
            {
                return null;
            }

            return TimeSpan.FromSeconds(double.Parse(queryParameterStringValue));
        }
    }
}

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

const timeout = "timeout";
const retryInterval = "retryInterval";

module.exports = async function (context, req) {
    const client = df.getClient(context);
    const instanceId = await client.startNew(req.params.functionName, undefined, req.body);

    context.log(`Started orchestration with ID = '${instanceId}'.`);

    const timeoutInMilliseconds = getTimeInSeconds(req, timeout) || 30000;
    const retryIntervalInMilliseconds = getTimeInSeconds(req, retryInterval) || 1000;

    const response = client.waitForCompletionOrCreateCheckStatusResponse(
        context.bindingData.req,
        instanceId,
        timeoutInMilliseconds,
        retryIntervalInMilliseconds
    );
    return response;
};

function getTimeInSeconds(req, queryParameterName) {
    const queryValue = req.query[queryParameterName];
    return queryValue
        ? queryValue * 1000 // expected to be in seconds
        : undefined;
}

A function.json konfigurációjának kezdőpéldányait lásd.

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

timeout = "timeout"
retry_interval = "retryInterval"

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, req.get_body())
    logging.log(f"Started orchestration with ID = '${instance_id}'.")

    timeout_in_milliseconds = get_time_in_seconds(req, timeout)
    timeout_in_milliseconds = timeout_in_milliseconds if timeout_in_milliseconds != None else 30000
    retry_interval_in_milliseconds = get_time_in_seconds(req, retry_interval)
    retry_interval_in_milliseconds = retry_interval_in_milliseconds if retry_interval_in_milliseconds != None else 1000

    return await client.wait_for_completion_or_create_check_status_response(
        req,
        instance_id,
        timeout_in_milliseconds,
        retry_interval_in_milliseconds
    )

def get_time_in_seconds(req: func.HttpRequest, query_parameter_name: str):
    query_value = req.params.get(query_parameter_name)
    return query_value if query_value != None else 1000

Java jelenleg nem rendelkezik egyetlen metódussal ehhez a forgatókönyvhöz, de néhány további kódsor használatával implementálhatja.

@FunctionName("HttpStartAndWait")
public HttpResponseMessage httpStartAndWait(
        @HttpTrigger(name = "req", route = "orchestrators/{functionName}/wait", methods = {HttpMethod.POST}) HttpRequestMessage<?> req,
        @DurableClientInput(name = "durableContext") DurableClientContext durableContext,
        @BindingName("functionName") String functionName,
        final ExecutionContext context) {

    DurableTaskClient client = durableContext.getClient();
    String instanceId = client.scheduleNewOrchestrationInstance(functionName);
    context.getLogger().info("Created new Java orchestration with instance ID = " + instanceId);
    try {
        String timeoutString = req.getQueryParameters().get("timeout");
        Integer timeoutInSeconds = Integer.parseInt(timeoutString);
        OrchestrationMetadata orchestration = client.waitForInstanceCompletion(
                instanceId,
                Duration.ofSeconds(timeoutInSeconds),
                true /* getInputsAndOutputs */);
        return req.createResponseBuilder(HttpStatus.OK)
                .body(orchestration.getSerializedOutput())
                .header("Content-Type", "application/json")
                .build();
    } catch (TimeoutException timeoutEx) {
        // timeout expired - return a 202 response
        return durableContext.createCheckStatusResponse(req, instanceId);
    }
}

A Durable Task SDK-k olyan módszert biztosítanak, amely megvárja, amíg a vezénylés szinkron módon befejeződik.

using Microsoft.DurableTask.Client;

// Wait for orchestration to complete with a timeout
OrchestrationMetadata metadata = await client.WaitForInstanceCompletionAsync(
    instanceId,
    timeout: TimeSpan.FromSeconds(30),
    getInputsAndOutputs: true);

if (metadata.RuntimeStatus == OrchestrationRuntimeStatus.Completed)
{
    Console.WriteLine($"Output: {metadata.SerializedOutput}");
}

from durabletask.azuremanaged.client import DurableTaskSchedulerClient

# Wait for orchestration to complete with a timeout
state = client.wait_for_orchestration_completion(
    instance_id,
    timeout=30,
    fetch_payloads=True)

if state.runtime_status == 'COMPLETED':
    print(f"Output: {state.serialized_output}")

import com.microsoft.durabletask.DurableTaskClient;
import com.microsoft.durabletask.OrchestrationMetadata;
import java.time.Duration;
import java.util.concurrent.TimeoutException;

// Wait for orchestration to complete with a timeout
try {
    OrchestrationMetadata metadata = client.waitForInstanceCompletion(
            instanceId,
            Duration.ofSeconds(30),
            true /* getInputsAndOutputs */);
    System.out.println("Output: " + metadata.getSerializedOutput());
} catch (TimeoutException e) {
    System.out.println("Orchestration did not complete within timeout");
}

import { createAzureManagedClient } from "@microsoft/durabletask-js-azuremanaged";
import { OrchestrationStatus } from "@microsoft/durabletask-js";

const client = createAzureManagedClient(connectionString);

// Wait for orchestration to complete with a timeout
const state = await client.waitForOrchestrationCompletion(instanceId, true, 30);

if (state?.runtimeStatus === OrchestrationStatus.COMPLETED) {
    console.log(`Output: ${state.serializedOutput}`);
}

Hívja meg a függvényt a következő sor használatával. Az időtúllépéshez használjon két másodpercet, az újrapróbálkozási időközhöz pedig 0,5 másodpercet:

curl -X POST "http://localhost:7071/orchestrators/E1_HelloSequence/wait?timeout=2&retryInterval=0.5"

Megjegyzés:

A fenti cURL-parancs feltételezi, hogy van egy E1_HelloSequence elnevezésű vezénylő függvény a projektjében. Az HTTP-eseményindító függvény megírásának módja lehetővé teszi, hogy bármelyik vezénylő függvény nevét felhasználja a projektben.

Az orchestration példány válaszának megérkezéséhez szükséges idő függvényében két lehetőség van:

Az orchestration példányok a megadott időkorláton belül fejeződnek be (ebben az esetben két másodperc), és a válasz az orchestration példány tényleges kimenete, amely szinkron módon kerül kézbesítésre.

HTTP/1.1 200 OK
Content-Type: application/json; charset=utf-8
Date: Thu, 14 Dec 2021 06:14:29 GMT
Transfer-Encoding: chunked

[
    "Hello Tokyo!",
    "Hello Seattle!",
    "Hello London!"
]

Orchestrációs példányok nem tudnak befejeződni a meghatározott időkorláton belül, és a válasz a HTTP API URL-felderítésben ismertetett alapértelmezett válasz:

HTTP/1.1 202 Accepted
Content-Type: application/json; charset=utf-8
Date: Thu, 14 Dec 2021 06:13:51 GMT
Location: http://localhost:7071/runtime/webhooks/durabletask/instances/d3b72dddefce4e758d92f4d411567177?taskHub={taskHub}&connection={connection}&code={systemKey}
Retry-After: 10
Transfer-Encoding: chunked

{
    "id": "d3b72dddefce4e758d92f4d411567177",
    "sendEventPostUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/d3b72dddefce4e758d92f4d411567177/raiseEvent/{eventName}?taskHub={taskHub}&connection={connection}&code={systemKey}",
    "statusQueryGetUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/d3b72dddefce4e758d92f4d411567177?taskHub={taskHub}&connection={connection}&code={systemKey}",
    "terminatePostUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/d3b72dddefce4e758d92f4d411567177/terminate?reason={text}&taskHub={taskHub}&connection={connection}&code={systemKey}",
    "suspendPostUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/d3b72dddefce4e758d92f4d411567177/suspend?reason={text}&taskHub={taskHub}&connection={connection}&code={systemKey}",
    "resumePostUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/d3b72dddefce4e758d92f4d411567177/resume?reason={text}&taskHub={taskHub}&connection={connection}&code={systemKey}"
}

Megjegyzés:

A webhook URL-címeinek formátuma eltérhet attól függően, hogy melyik verziójú Azure Functions-gazdagépet futtatja. Az előző példa az Azure Functions 3.0-gazdagépre mutat.

HTTP-felügyeleti webhook URL-címek lekérése vezénylési példányokhoz

Külső rendszer használata események figyelésére vagy vezénylésére. A külső rendszerek a Durable Functions webhook URL-címein keresztül kommunikálnak, amelyek a HTTP API URL feltárásakor ismertetett alapértelmezett válasz részét képezik. A webhook URL-címei programozott módon is hozzáférhetők az orchestrációs klienskötés használatával. A HTTP kezelési hasznos adatok létrehozása API egy szerializálható objektumot kap, mely ezeket a webhook URL-címeket tartalmazza.

Az HTTP felügyeleti terhelés létrehozása API egy paraméterrel rendelkezik:

Példányazonosító: A példány egyedi azonosítója.

A metódusok a következő sztringtulajdonságokkal rendelkező objektumot adnak vissza:

Azonosító: Az orkesztráció példányazonosítója (a bemenettel InstanceId megegyezőnek kell lennie).
StatusQueryGetUri: Az orchestrációs példány állapot URL-címe.
SendEventPostUri: A vezénylési példány "esemény létrehozása" URL-címe.
TerminatePostUri: A vezénylési példány "befejezés" URL-címe.
PurgeHistoryDeleteUri: Az orchestration példány "törlési előzményeinek" URL-címe.
SuspendPostUri: Az orkesztációs példány "felfüggesztés" URL-címe.
ResumePostUri: A vezénylési példány "folytatás" URL-címe.

A függvények az objektumok példányait külső rendszereknek küldik, hogy figyeljék az eseményeket, vagy eseményeket generáljanak a megfelelő orchestrationökön, ahogyan az alábbi példákban is látható.

[FunctionName("SendInstanceInfo")]
public static void SendInstanceInfo(
    [ActivityTrigger] IDurableActivityContext ctx,
    [DurableClient] IDurableOrchestrationClient client,
    [CosmosDB(
        databaseName: "MonitorDB",
        containerName: "HttpManagementPayloads",
        Connection = "CosmosDBConnectionSetting")]out dynamic document)
{
    HttpManagementPayload payload = client.CreateHttpManagementPayload(ctx.InstanceId);

    // send the payload to Azure Cosmos DB
    document = new { Payload = payload, id = ctx.InstanceId };
}

Megjegyzés:

Az előző C#-kód a folyamaton belüli modellt IDurableOrchestrationClientIDurableActivityContexthasználja, és a Durable Functions bővítmény újabb verzióiban elavultként van megjelölve. Új .NET-projektek esetén fontolja meg a .NET izolált feldolgozói modellDurableTaskClienthasználatát. További információkért tekintse meg a Durable Functions-verziókról szóló cikket.

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

modules.exports = async function(context, ctx) {
    const client = df.getClient(context);

    const payload = client.createHttpManagementPayload(ctx.instanceId);

    // send the payload to Azure Cosmos DB
    context.bindings.document = JSON.stringify({
        id: ctx.instanceId,
        payload,
    });
};

A function.json konfigurációjának kezdőpéldányait lásd.

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

async def main(req: func.HttpRequest, starter: str, instance_id: str) -> func.cosmosdb.cdb.Document:
    client = df.DurableOrchestrationClient(starter)

    payload = client.create_check_status_response(req, instance_id).get_body().decode()

    return func.cosmosdb.CosmosDBConverter.encode({
        id: instance_id,
        payload: payload
    })

using namespace System.Net

param($Request, $TriggerMetadata)

$InstanceId = $Request.Body.InstanceId
$Response = New-DurableOrchestrationCheckStatusResponse -Request $Request -InstanceId $InstanceId
Push-OutputBinding -Name Response -Value $Response

Az orchestration példányok visszatekerése

Ha a vezénylési folyamat váratlan okok miatt meghibásodik, a példányt egy erre a célra létrehozott API használatával visszatekerheti egy korábbi hibátlan állapotba.

Megjegyzés:

Ez az API nem helyettesíti a megfelelő hibakezelési és újrapróbálkozési szabályzatokat. Ehelyett csak olyan esetekben használható, amikor a vezénylési példányok váratlan okokból meghiúsulnak. A nem Failed (például Running, , Pending, Terminatedvagy Completed) állapotban lévő vezénylések nem lehetnek "újrahangolva". A hibakezelési és újrapróbálkozési szabályzatokkal kapcsolatos további információkért tekintse meg a hibakezelési cikket.

Használja az RewindAsyncrewind (.NET) vagy (JavaScript) metódusát, hogy az orchestrációt visszahelyezze a Futtatás állapotba. Ez a módszer a vezénylési hibát okozó tevékenység- vagy alvezénylési végrehajtási hibákat is újra futtatja.

Tegyük fel például, hogy egy munkafolyamat több emberi jóváhagyást is magában foglal. Tegyük fel, hogy tevékenységfüggvények sorozata értesíti valakit arról, hogy jóváhagyásra van szükség, és várja meg a valós idejű választ. Miután az összes jóváhagyási tevékenység kap válaszokat vagy időtúllépésre kerül, tegyük fel, hogy egy másik tevékenység meghiúsul egy alkalmazás helytelen konfigurációból adódóan, például egy érvénytelen adatbázis kapcsolati karakterlánc miatt. Az eredmény egy orchestrációs hiba a folyamat mélyén. A RewindAsync (.NET) vagy rewind (JavaScript) API-val az alkalmazás rendszergazdája kijavíthatja a konfigurációs hibát, és visszatekerheti a sikertelen vezénylést közvetlenül a hiba előtt. Az emberi beavatkozás egyik lépését sem kell újra jóváhagyni, és a vezénylés sikeresen befejeződhet.

Megjegyzés:

A visszatekerés funkció nem támogatja a tartós időzítőket használó vezénylési példányok visszatekerését.

[FunctionName("RewindInstance")]
public static Task Run(
    [DurableClient] IDurableOrchestrationClient client,
    [QueueTrigger("rewind-queue")] string instanceId)
{
    string reason = "Orchestrator failed and needs to be revived.";
    return client.RewindAsync(instanceId, reason);
}

Megjegyzés:

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

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

    const reason = "Orchestrator failed and needs to be revived.";
    return client.rewind(instanceId, reason);
};

A function.json konfigurációjának kezdőpéldányait lásd.

using Microsoft.DurableTask.Client;

string reason = "Orchestrator failed and needs to be revived.";
await client.RewindInstanceAsync(instanceId, reason);

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

const client = createAzureManagedClient(connectionString);

const reason = "Orchestrator failed and needs to be revived.";
await client.rewindInstance(instanceId, reason);

Orchestration példányok újraindítása

A vezénylés újraindítása új példányt hoz létre egy korábban futtatott példány előzményei alapján. Ez a funkció akkor hasznos, ha ugyanazzal a bemeneti és példányazonosító-mintával szeretne újrakezdeni egy orchestration-t, az eredeti alapján új futtatást létrehozva.

[FunctionName("RestartInstance")]
public static Task Run(
    [DurableClient] IDurableOrchestrationClient client,
    [QueueTrigger("restart-queue")] string instanceId)
{
    return client.RestartAsync(instanceId, restartWithNewInstanceId: true);
}

Megjegyzés:

using Microsoft.DurableTask.Client;

// Restart an orchestration with a new instance ID
string newInstanceId = await client.RestartInstanceAsync(instanceId, restartWithNewInstanceId: true);
Console.WriteLine($"Restarted as new instance: {newInstanceId}");

// Restart an orchestration keeping the same instance ID
await client.RestartInstanceAsync(instanceId, restartWithNewInstanceId: false);

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

const client = createAzureManagedClient(connectionString);

// Restart an orchestration with a new instance ID
const newInstanceId = await client.restartOrchestration(instanceId, true);
console.log(`Restarted as new instance: ${newInstanceId}`);

// Restart an orchestration keeping the same instance ID
await client.restartOrchestration(instanceId, false);

Orchestration példány előzményeinek törlése

Az orchestrationhoz kapcsolt összes adat eltávolításához törölje az instance előzményeket. Törölje például a befejezett példányhoz társított tárerőforrásokat. Használja az orchestration kliens által definiált purge instance API-t.

Az alábbi példa bemutatja, hogyan törölhet egy orchestration példányt.

[FunctionName("PurgeInstanceHistory")]
public static Task Run(
    [DurableClient] IDurableOrchestrationClient client,
    [QueueTrigger("purge-queue")] string instanceId)
{
    return client.PurgeInstanceHistoryAsync(instanceId);
}

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

module.exports = async function(context, instanceId) {
    const client = df.getClient(context);
    return client.purgeInstanceHistory(instanceId);
};

A function.json konfigurációjának kezdőpéldányait lásd.

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

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

    return await client.purge_instance_history(instance_id)

@FunctionName("PurgeInstance")
public HttpResponseMessage purgeInstance(
        @HttpTrigger(name = "req", methods = {HttpMethod.POST}, route = "purge/{instanceID}") HttpRequestMessage<?> req,
        @DurableClientInput(name = "durableContext") DurableClientContext durableContext,
        @BindingName("instanceID") String instanceID) {
    PurgeResult result = durableContext.getClient().purgeInstance(instanceID);
    if (result.getDeletedInstanceCount() == 0) {
        return req.createResponseBuilder(HttpStatus.NOT_FOUND)
                .body("No completed instance with ID '" + instanceID + "' was found!")
                .build();
    } else {
        return req.createResponseBuilder(HttpStatus.OK)
                .body("Successfully purged data for " + instanceID)
                .build();
    }
}

using Microsoft.DurableTask.Client;

// Purge a single orchestration instance
PurgeResult result = await client.PurgeInstanceAsync(instanceId);
Console.WriteLine($"Purged {result.PurgedInstanceCount} instance(s).");

from durabletask.azuremanaged.client import DurableTaskSchedulerClient

# Purge a single orchestration instance
result = client.purge_orchestration(instance_id)
print(f"Purged {result.deleted_instance_count} instance(s).")

import com.microsoft.durabletask.DurableTaskClient;
import com.microsoft.durabletask.PurgeResult;

// Purge a single orchestration instance
PurgeResult result = client.purgeInstance(instanceId);
System.out.println("Purged " + result.getDeletedInstanceCount() + " instance(s).");

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

const client = createAzureManagedClient(connectionString);

// Purge a single orchestration instance
const result = await client.purgeOrchestration(instanceId);
console.log(`Purged ${result?.deletedInstanceCount ?? 0} instance(s).`);

Az alábbi példa egy időzítő által aktivált függvényt mutat be, amely törli azon vezénylési példányok előzményeit, amelyek a megadott időintervallum után fejeződtek be. Ebben az esetben eltávolítja a 30 vagy több nappal ezelőtt befejezett összes példány adatait. Ez a példafüggvény naponta egyszer, UTC 12:00-kor lesz futtatva:

[FunctionName("PurgeInstanceHistory")]
public static Task Run(
    [DurableClient] IDurableOrchestrationClient client,
    [TimerTrigger("0 0 12 * * *")] TimerInfo myTimer)
{
    return client.PurgeInstanceHistoryAsync(
        DateTime.MinValue,
        DateTime.UtcNow.AddDays(-30),  
        new List<OrchestrationStatus>
        {
            OrchestrationStatus.Completed
        });
}

Megjegyzés:

A purgeInstanceHistoryBy metódus több példány példányelőzményeinek feltételes törlésére használható.

function.json

{
  "bindings": [
    {
      "schedule": "0 0 12 * * *",
      "name": "myTimer",
      "type": "timerTrigger",
      "direction": "in"
    },
    {
      "name": "starter",
      "type": "durableClient",
      "direction": "in"
    }
  ],
  "disabled": false
}

Megjegyzés:

Ez a példa a Durable Functions 2.x-es verzióját célozza meg. Használja az orchestrationClient-t a durableClient helyett az 1.x verzióban.

index.js

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

module.exports = async function (context, myTimer) {
    const client = df.getClient(context);
    const createdTimeFrom = new Date(0);
    const createdTimeTo = new Date().setDate(today.getDate() - 30);
    const runtimeStatuses = [ df.OrchestrationRuntimeStatus.Completed ];
    return client.purgeInstanceHistoryBy(createdTimeFrom, createdTimeTo, runtimeStatuses);
};

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

async def main(req: func.HttpRequest, starter: str, instance_id: str) -> func.HttpResponse:
    client = df.DurableOrchestrationClient(starter)
    created_time_from = datetime.min
    created_time_to = datetime.today() + timedelta(days = -30)
    runtime_statuses = [OrchestrationRuntimeStatus.Completed]

    return await client.purge_instance_history_by(created_time_from, created_time_to, runtime_statuses)

@FunctionName("PurgeInstances")
public void purgeInstances(
        @TimerTrigger(name = "purgeTimer", schedule = "0 0 12 * * *") String timerInfo,
        @DurableClientInput(name = "durableContext") DurableClientContext durableContext,
        ExecutionContext context) throws TimeoutException {
    PurgeInstanceCriteria criteria = new PurgeInstanceCriteria()
            .setCreatedTimeFrom(Instant.now().minus(Duration.ofDays(60)))
            .setCreatedTimeTo(Instant.now().minus(Duration.ofDays(30)))
            .setRuntimeStatusList(List.of(OrchestrationRuntimeStatus.COMPLETED));
    PurgeResult result = durableContext.getClient().purgeInstances(criteria);
    context.getLogger().info(String.format("Purged %d instance(s)", result.getDeletedInstanceCount()));
}

using Microsoft.DurableTask.Client;

// Purge completed instances older than 30 days
var filter = new PurgeInstancesFilter(
    CreatedFrom: DateTime.MinValue,
    CreatedTo: DateTime.UtcNow.AddDays(-30),
    Statuses: new[] { OrchestrationRuntimeStatus.Completed });

PurgeResult result = await client.PurgeAllInstancesAsync(filter);
Console.WriteLine($"Purged {result.PurgedInstanceCount} instance(s).");

from durabletask.azuremanaged.client import DurableTaskSchedulerClient
from datetime import datetime, timedelta

# Purge completed instances older than 30 days
result = client.purge_orchestrations(
    created_time_from=datetime.min,
    created_time_to=datetime.utcnow() - timedelta(days=30),
    runtime_status=['COMPLETED']
)
print(f"Purged {result.deleted_instance_count} instance(s).")

import com.microsoft.durabletask.DurableTaskClient;
import com.microsoft.durabletask.PurgeInstanceCriteria;
import com.microsoft.durabletask.PurgeResult;
import com.microsoft.durabletask.OrchestrationRuntimeStatus;
import java.time.Duration;
import java.time.Instant;
import java.util.List;

// Purge completed instances older than 30 days
PurgeInstanceCriteria criteria = new PurgeInstanceCriteria()
    .setCreatedTimeTo(Instant.now().minus(Duration.ofDays(30)))
    .setRuntimeStatusList(List.of(OrchestrationRuntimeStatus.COMPLETED));

PurgeResult result = client.purgeInstances(criteria);
System.out.println("Purged " + result.getDeletedInstanceCount() + " instance(s).");

import { createAzureManagedClient } from "@microsoft/durabletask-js-azuremanaged";
import { OrchestrationStatus, PurgeInstanceCriteria } from "@microsoft/durabletask-js";

const client = createAzureManagedClient(connectionString);

// Purge completed instances older than 30 days
const thirtyDaysAgo = new Date(Date.now() - 30 * 24 * 60 * 60 * 1000);

const criteria = new PurgeInstanceCriteria();
criteria.setCreatedTimeTo(thirtyDaysAgo);
criteria.setRuntimeStatusList([OrchestrationStatus.COMPLETED]);

const result = await client.purgeOrchestration(criteria);
console.log(`Purged ${result?.deletedInstanceCount ?? 0} instance(s).`);

Megjegyzés:

Ahhoz, hogy a törlési előzményművelet sikeres legyen, a célpéldány futtatókörnyezeti állapotának befejezettnek, megszakítottnak vagy sikertelennek kell lennie.

Következő lépések

Verziószámozás kezelése

Beépített HTTP API-referencia példánykezeléshez

A Durable Task SDK-k használatának első lépései

Tudnivalók a Durable Task Schedulerről

Visszajelzés

Hasznosnak találta ezt az oldalt?

Last updated on 2026-04-22

Orchestration példányok kezelése

Példányok indítása

Lekérdezéspéldányok

Az összes orchestration példány lekérdezése

Orchestration példányok lekérdezése szűrőkkel

Orchesztrációs példányok megszüntetése

Orkesztációs példányok felfüggesztése és újraindítása

Események küldése példányokra

Várakozás a vezénylés befejezésére

HTTP-felügyeleti webhook URL-címek lekérése vezénylési példányokhoz

Az orchestration példányok visszatekerése

Orchestration példányok újraindítása

Orchestration példány előzményeinek törlése

Következő lépések

Visszajelzés

További források