Оркестраторы "синглтон"

Для фоновых заданий часто необходимо убедиться, что одновременно запускается только одна инстанция конкретного оркестратора. Это поведение можно обеспечить в Durable Functions или Durable Task SDKs путем назначения определенного идентификатора экземпляра оркестратору при его создании, и проверки, запущен ли уже экземпляр с данным идентификатором, прежде чем запускать новый экземпляр.

Это важно

В настоящее время пакет SDK для устойчивых задач PowerShell недоступен.

Пример Singleton

В следующем примере показана функция HTTP-триггера, которая создает оркестрацию одиночных фоновых задач. Код пытается убедиться, что для указанного идентификатора экземпляра существует только один активный экземпляр.

C#

[Function("HttpStartSingle")]
public static async Task<HttpResponseData> RunSingle(
    [HttpTrigger(AuthorizationLevel.Function, "post", Route = "orchestrators/{functionName}/{instanceId}")] HttpRequestData req,
    [DurableClient] DurableTaskClient starter,
    string functionName,
    string instanceId,
    FunctionContext executionContext)
{
    ILogger logger = executionContext.GetLogger("HttpStartSingle");

    // Check if an instance with the specified ID already exists or an existing one stopped running(completed/failed/terminated).
    OrchestrationMetadata? existingInstance = await starter.GetInstanceAsync(instanceId, getInputsAndOutputs: false);
    if (existingInstance == null 
    || existingInstance.RuntimeStatus == OrchestrationRuntimeStatus.Completed 
    || existingInstance.RuntimeStatus == OrchestrationRuntimeStatus.Failed 
    || existingInstance.RuntimeStatus == OrchestrationRuntimeStatus.Terminated)
    {
        // An instance with the specified ID doesn't exist or an existing one stopped running, create one.
        string requestBody = await new StreamReader(req.Body).ReadToEndAsync();
        await starter.ScheduleNewOrchestrationInstanceAsync(functionName, requestBody, new StartOrchestrationOptions { InstanceId = instanceId });
        logger.LogInformation($"Started orchestration with ID = '{instanceId}'.");
        return await starter.CreateCheckStatusResponseAsync(req, instanceId);
    }
    else
    {
        // An instance with the specified ID exists or an existing one still running, don't create one.
        var response = req.CreateResponse(HttpStatusCode.Conflict);
        await response.WriteStringAsync($"An instance with ID '{instanceId}' already exists.");
        return response;
    }
}

Замечание

Предыдущий код C# предназначен для изолированной рабочей модели, которая является рекомендуемой моделью для приложений .NET. Дополнительные сведения о различиях между моделями рабочих процессов в процессе и изолированными моделями см. в статье о версиях Durable Functions.

JavaScript

Замечание

В этом примере JavaScript используется модель программирования Node.js версии 3, которая использует function.json. Если вы используете модель программирования Node.js версии 4, используйте вместо этого шаблон кода версии 4.

function.json

{
  "bindings": [
    {
      "authLevel": "function",
      "name": "req",
      "type": "httpTrigger",
      "direction": "in",
      "route": "orchestrators/{functionName}/{instanceId}",
      "methods": ["post"]
    },
    {
      "name": "starter",
      "type": "orchestrationClient",
      "direction": "in"
    },
    {
      "name": "$return",
      "type": "http",
      "direction": "out"
    }
  ]
}

index.js

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

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

    const instanceId = req.params.instanceId;
    const functionName = req.params.functionName;

    // Check if an instance with the specified ID already exists or an existing one stopped running(completed/failed/terminated).
    const existingInstance = await client.getStatus(instanceId);
    if (!existingInstance 
        || existingInstance.runtimeStatus == "Completed" 
        || existingInstance.runtimeStatus == "Failed" 
        || existingInstance.runtimeStatus == "Terminated") {
        // An instance with the specified ID doesn't exist or an existing one stopped running, create one.
        const eventData = req.body;
        await client.startNew(functionName, instanceId, eventData);
        context.log(`Started orchestration with ID = '${instanceId}'.`);
        return client.createCheckStatusResponse(req, instanceId);
    } else {
        // An instance with the specified ID exists or an existing one still running, don't create one.
        return {
            status: 409,
            body: `An instance with ID '${instanceId}' already exists.`,
        };
    }
};

Python

function_app.py

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

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

@app.route(route="orchestrators/{functionName}/{instanceId}", methods=["POST"])
@app.durable_client_input(client_name="client")
async def http_start_single(req: func.HttpRequest, client):
        instance_id = req.route_params["instanceId"]
        function_name = req.route_params["functionName"]

    existing_instance = await client.get_status(instance_id)

        if not existing_instance or existing_instance.runtime_status in [
                df.OrchestrationRuntimeStatus.Completed,
                df.OrchestrationRuntimeStatus.Failed,
                df.OrchestrationRuntimeStatus.Terminated,
        ]:
        event_data = req.get_body()
                instance_id = await client.start_new(function_name, instance_id, event_data)
        logging.info(f"Started orchestration with ID = '{instance_id}'.")
        return client.create_check_status_response(req, instance_id)

        return func.HttpResponse(
                body=f"An instance with ID '{instance_id}' already exists.",
                status_code=409,
        )

PowerShell

using namespace System.Net

param($Request, $TriggerMetadata)

$FunctionName = $Request.Params.FunctionName
$InstanceId = $Request.Params.InstanceId

# Check if an instance with the specified ID already exists
$existingInstance = Get-DurableStatus -InstanceId $InstanceId

if (-not $existingInstance -or 
    $existingInstance.RuntimeStatus -eq "Completed" -or 
    $existingInstance.RuntimeStatus -eq "Failed" -or 
    $existingInstance.RuntimeStatus -eq "Terminated") {

    # An instance with the specified ID doesn't exist or stopped running, create one
    $InstanceId = Start-DurableOrchestration -FunctionName $FunctionName -InstanceId $InstanceId -Input $Request.Body
    Write-Host "Started orchestration with ID = '$InstanceId'."

    $Response = New-DurableOrchestrationCheckStatusResponse -Request $Request -InstanceId $InstanceId
    Push-OutputBinding -Name Response -Value $Response
}
else {
    # An instance with the specified ID exists or still running
    Push-OutputBinding -Name Response -Value ([HttpResponseContext]@{
        StatusCode = [HttpStatusCode]::Conflict
        Body = "An instance with ID '$InstanceId' already exists."
    })
}

Java

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

    String instanceID = "StaticID";
    DurableTaskClient client = durableContext.getClient();

    // Check to see if an instance with this ID is already running
    OrchestrationMetadata metadata = client.getInstanceMetadata(instanceID, false);
    if (metadata.isRunning()) {
        return req.createResponseBuilder(HttpStatus.CONFLICT)
                .body("An instance with ID '" + instanceID + "' already exists.")
                .build();
    }

    // No such instance exists - create a new one. De-dupe is handled automatically
    // in the storage layer if another function tries to also use this instance ID.
    client.scheduleNewOrchestrationInstance("MyOrchestration", null, instanceID);
    return durableContext.createCheckStatusResponse(req, instanceID);
}

В следующем примере показано, как создать однотонную оркестрацию с помощью пакетов SDK для устойчивых задач. Код пытается убедиться, что для указанного идентификатора экземпляра существует только один активный экземпляр.

C#

using Microsoft.DurableTask.Client;

// Check if an instance with the specified ID already exists
string instanceId = "singleton-job";
OrchestrationMetadata? existingInstance = await client.GetInstanceAsync(instanceId, getInputsAndOutputs: false);

if (existingInstance == null ||
    existingInstance.RuntimeStatus == OrchestrationRuntimeStatus.Completed ||
    existingInstance.RuntimeStatus == OrchestrationRuntimeStatus.Failed ||
    existingInstance.RuntimeStatus == OrchestrationRuntimeStatus.Terminated)
{
    // An instance with the specified ID doesn't exist or an existing one stopped running, create one.
    await client.ScheduleNewOrchestrationInstanceAsync("MyOrchestration", input, new StartOrchestrationOptions(instanceId));
    Console.WriteLine($"Started orchestration with ID = '{instanceId}'.");
}
else
{
    // An instance with the specified ID exists or an existing one still running.
    Console.WriteLine($"An instance with ID '{instanceId}' already exists.");
}

Python

from durabletask.azuremanaged.client import DurableTaskSchedulerClient

instance_id = "singleton-job"

# Check if an instance with the specified ID already exists
existing_instance = client.get_orchestration_state(instance_id)

if (existing_instance is None or
    existing_instance.runtime_status in ['COMPLETED', 'FAILED', 'TERMINATED']):
    # An instance with the specified ID doesn't exist or an existing one stopped running, create one.
    client.schedule_new_orchestration(my_orchestration, input=input_data, instance_id=instance_id)
    print(f"Started orchestration with ID = '{instance_id}'.")
else:
    # An instance with the specified ID exists or an existing one still running.
    print(f"An instance with ID '{instance_id}' already exists.")

Java

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

String instanceId = "singleton-job";

// Check to see if an instance with this ID is already running
OrchestrationMetadata existingInstance = client.getInstanceMetadata(instanceId, false);

if (existingInstance == null || !existingInstance.isRunning()) {
    // An instance doesn't exist or finished - create one
    client.scheduleNewOrchestrationInstance("MyOrchestration", input, instanceId);
    System.out.println("Started orchestration with ID = '" + instanceId + "'.");
} else {
    // An instance with the specified ID exists and is still running
    System.out.println("An instance with ID '" + instanceId + "' already exists.");
}

JavaScript

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

const client = createAzureManagedClient(connectionString);

const instanceId = "singleton-job";

// Check if an instance with the specified ID already exists
const existingInstance = await client.getOrchestrationState(instanceId, false);

if (!existingInstance ||
    existingInstance.runtimeStatus === OrchestrationStatus.COMPLETED ||
    existingInstance.runtimeStatus === OrchestrationStatus.FAILED ||
    existingInstance.runtimeStatus === OrchestrationStatus.TERMINATED) {
    // An instance with the specified ID doesn't exist or an existing one stopped running, create one.
    await client.scheduleNewOrchestration("MyOrchestration", input, instanceId);
    console.log(`Started orchestration with ID = '${instanceId}'.`);
} else {
    // An instance with the specified ID exists or an existing one still running.
    console.log(`An instance with ID '${instanceId}' already exists.`);
}

PowerShell

Пакет SDK для устойчивых задач недоступен для PowerShell. Вместо этого используйте Durable Functions.

По умолчанию идентификаторы экземпляров создаются в виде случайных GUID. Однако в предыдущем примере используется определенный идентификатор экземпляра. Затем код получает метаданные экземпляра оркестрационного процесса, чтобы проверить, запущен ли экземпляр с указанным идентификатором. Если такой экземпляр не запущен, создается новый экземпляр с таким идентификатором.

Замечание

В этом примере существует потенциальное конкурентное состояние. Если два экземпляра выполняются одновременно, оба вызова могут сообщать об успешном выполнении, но фактически запускается только один экземпляр оркестрации. В зависимости от ваших требований это может иметь нежелательные побочные эффекты.

Сведения о реализации функции оркестратора на самом деле не имеют значения. Это может быть обычная функция оркестратора, которая запускается и завершается, или та, которая выполняется постоянно (т. е. вечная оркестрация). Важно отметить, что одновременно работает только один экземпляр.

Дальнейшие действия

Узнайте о родных возможностях HTTP для оркестровок

Начало работы с Durable Task SDK