对于后台作业,你通常需要确保特定业务流程协调程序同一时间只运行一个实例,以防止重复的协调并发运行。 可以在 Durable Functions 或 Durable Task SDKs 中实现单一实例模式,方法是在创建时将特定的实例 ID 分配给业务流程协调程序,然后检查具有该 ID 的实例是否已经在运行,再启动新的实例。
本文展示如何通过每种支持语言的代码示例实现单例协调器。
先决条件
注释
单例模式存在潜在的竞态条件。 如果两个客户端同时执行检查和启动逻辑,则两个调用可能会报告成功,但实际上只有一个业务流程实例启动。 根据你的要求,这可能会产生不良的副作用。 如果需要严格的单实例保证,请考虑添加其他锁定机制。
重要
目前,PowerShell Durable Task SDK 不可用。
单例业务流程协调程序示例
以下示例演示了一个 HTTP 触发器函数,该函数创建单例后台作业协调。 代码尝试确保指定实例 ID 仅存在一个活动实例。
[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;
}
}
重要
此 JavaScript 示例使用 Node.js 编程模型 v3,该模型使用 function.json。 如果使用 Node.js 编程模型 v4 (当前默认值),请改用 v4 代码优先模式。
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.`,
};
}
};
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,
)
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."
})
}
@FunctionName("HttpStartSingle")
public HttpResponseMessage runSingle(
@HttpTrigger(name = "req", route = "orchestrators/{functionName}/{instanceId}") HttpRequestMessage<?> req,
@DurableClientInput(name = "durableContext") DurableClientContext durableContext) {
String instanceId = req.getHeaders().getOrDefault("instanceId", "singleton-job");
DurableTaskClient client = durableContext.getClient();
// Check to see if an instance with this ID is already running
OrchestrationMetadata metadata = client.getInstanceMetadata(instanceId, false);
if (metadata == null || !metadata.isRunning()) {
// No such instance exists or it finished - 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);
}
return req.createResponseBuilder(HttpStatus.CONFLICT)
.body("An instance with ID '" + instanceId + "' already exists.")
.build();
}
以下示例演示如何使用持久任务 SDK 创建单例编排。 代码尝试确保指定实例 ID 仅存在一个活动实例。
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.");
}
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.")
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.");
}
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.`);
}
单例模式的工作原理
由于实例 ID 在任务中心内是唯一的,因此计划具有已知、固定 ID 的业务流程,并首先检查其状态会阻止重复的并发运行。 默认情况下,实例 ID 是随机生成的 GUID。 但是,在前面的示例中,将传入特定的实例 ID。 然后,该代码提取业务流程实例元数据,以检查具有该 ID 的实例是否已运行。 如果未运行此类实例,则会使用该 ID 创建新实例。
协调器函数本身可以使用任何模式:启动和完成的标准函数,或持续运行的 永久协调。 单一实例模式仅控制并发运行的实例数。
后续步骤