Singleton-orchestrators in Durable Functions (Azure Functions)

Voor achtergrondtaken moet u er vaak voor zorgen dat er slechts één exemplaar van een bepaalde orchestrator tegelijk wordt uitgevoerd. U kunt dit soort singletongedrag in Durable Functions controleren door een specifieke instantie-id toe te wijzen aan een orchestrator wanneer u deze maakt.

Singleton-voorbeeld

In het volgende voorbeeld ziet u een HTTP-triggerfunctie waarmee een singleton-achtergrondtaakindeling wordt gemaakt. De code zorgt ervoor dat er slechts één exemplaar bestaat voor een opgegeven exemplaar-id.

C#

[FunctionName("HttpStartSingle")]
public static async Task<HttpResponseMessage> RunSingle(
    [HttpTrigger(AuthorizationLevel.Function, methods: "post", Route = "orchestrators/{functionName}/{instanceId}")] HttpRequestMessage req,
    [DurableClient] IDurableOrchestrationClient starter,
    string functionName,
    string instanceId,
    ILogger log)
{
    // Check if an instance with the specified ID already exists or an existing one stopped running(completed/failed/terminated).
    var existingInstance = await starter.GetStatusAsync(instanceId);
    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.
        dynamic eventData = await req.Content.ReadAsAsync<object>();
        await starter.StartNewAsync(functionName, instanceId, eventData);
        log.LogInformation($"Started orchestration with ID = '{instanceId}'.");
        return starter.CreateCheckStatusResponse(req, instanceId);
    }
    else
    {
        // An instance with the specified ID exists or an existing one still running, don't create one.
        return new HttpResponseMessage(HttpStatusCode.Conflict)
        {
            Content = new StringContent($"An instance with ID '{instanceId}' already exists."),
        };
    }
}

Notitie

De vorige C#-code is voor Durable Functions 2.x. Voor Durable Functions 1.x moet u het kenmerk gebruiken OrchestrationClient in plaats van het DurableClient kenmerk en moet u het DurableOrchestrationClient parametertype gebruiken in plaats van IDurableOrchestrationClient. Zie het artikel Durable Functions versies voor meer informatie over de verschillen tussen versies.

JavaScript

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.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"
    }
  ]
}

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 = req.route_params['instanceId']
    function_name = req.route_params['functionName']

    existing_instance = await client.get_status(instance_id)

    if existing_instance.runtime_status in [df.OrchestrationRuntimeStatus.Completed, df.OrchestrationRuntimeStatus.Failed, df.OrchestrationRuntimeStatus.Terminated, None]:
        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)
    else:
        return {
            'status': 409,
            'body': f"An instance with ID '${existing_instance.instance_id}' 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);
}

Instantie-id's zijn standaard willekeurig gegenereerde GUID's. In het vorige voorbeeld wordt de instantie-id echter doorgegeven in routegegevens van de URL. De code haalt vervolgens de metagegevens van het indelingsexemplaar op om te controleren of een exemplaar met de opgegeven id al wordt uitgevoerd. Als er geen dergelijk exemplaar wordt uitgevoerd, wordt er een nieuw exemplaar gemaakt met die id.

Notitie

Er is een mogelijke racevoorwaarde in dit voorbeeld. Als twee exemplaren van HttpStartSingle gelijktijdig worden uitgevoerd, melden beide functieaanroepen dat het is gelukt, maar wordt slechts één indelingsexemplaar daadwerkelijk gestart. Afhankelijk van uw behoeften kan dit ongewenste bijwerkingen hebben.

De implementatiedetails van de orchestratorfunctie zijn eigenlijk niet van belang. Het kan een reguliere orchestratorfunctie zijn die wordt gestart en voltooid, of een functie die voor altijd wordt uitgevoerd (dat wil gezegd, een eeuwige indeling). Het belangrijke punt is dat er maar één exemplaar tegelijk wordt uitgevoerd.

Volgende stappen

Meer informatie over de systeemeigen HTTP-functies van indelingen