Durable Functions (Azure Functions) 中的永久性協調流程

文章
06/01/2023

「永久協調流程」是指永不結束的協調流程函式。想要在彙總工具和任何需要無限迴圈的情節下使用 Durable Functions 時，永久性協調流程就很有用。

協調流程記錄

如協調流程記錄主題中所述，「永久性工作架構」會持續追蹤每個函式協調流程的記錄。只要協調器函式繼續排程新的工作，此記錄就會持續成長。如果協調器函式進入無限迴圈，並持續排程工作，此記錄可能會變得非常大，而造成明顯的效能問題。「永久協調流程」概念是設計讓需要無限迴圈的應用程式減少發生這類問題。

重設和重新啟動

協調器函式會呼叫協調流程觸發程序繫結的 continue-as-new 方法來重設其狀態，而不會使用無限迴圈。此方法接受 JSON 序列化參數，此參數會變成新的輸入來產生下一個協調器函式。

呼叫 continue-as-new 時，協調流程執行個體會以新的輸入值重新啟動本身。相同的執行個體識別碼會保留下來，但協調器函式的記錄會重設。

注意

「永久性工作架構」會維護相同的執行個體識別碼，但在內部會為 continue-as-new 所重設的協調器函式建立新的「執行識別碼」。此執行識別碼不對外公開，但可能很助於了解偵錯協調流程執行。

定期工作範例

需要無限期地執行定期工作的程式碼，即為永久協調流程的一個使用案例。

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

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

    context.ContinueAsNew(null);
}

注意

先前的 C# 範例適用於 Durable Functions 2.x。針對 Durable Functions 1.x，您必須使用 DurableOrchestrationContext 而不是 IDurableOrchestrationContext。如需版本差異的詳細資訊，請參閱 Durable Functions 版本一文。

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

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

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

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

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

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

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

    context.continue_as_new(None)

main = df.Orchestrator.create(orchestrator_function)

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

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

    ctx.continueAsNew(null);
}

此範例與計時器觸發函式之間的差異在於，此處的清理觸發程序時間不是根據排程。例如，每小時執行函式的 CRON 排程會在 1:00、2:00、3:00 等時間執行它，很可能會陷入重疊問題。不過，在此範例中，如果清除需要 30 分鐘，則會排程在 1:00、2:30、4:00 等時間，不會有機會重疊。

啟動永久協調流程

使用 start-new 或 schedule-new 耐久型用戶端方法來啟動永久協調流程，就像您在使用任何其他協調流程函式時一樣。

注意

如果您需要確保單一永久協調流程正在執行，請務必在啟動協調流程時維持相同的執行個體 id。如需詳細資訊，請參閱執行個體管理。

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

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

注意

先前的程式碼適用於 Durable Functions 2.x。針對 Durable Functions 1.x，您必須使用 OrchestrationClient 屬性 (而非 DurableClient 屬性)，而且必須使用 DurableOrchestrationClient 參數類型 (而非 IDurableOrchestrationClient)。如需版本差異的詳細資訊，請參閱 Durable Functions 版本一文。

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

module.exports = async function (context, req) {
    const client = df.getClient(context);
    const instanceId = "StaticId";
    
    // null is used as the input, since there is no input in "Periodic_Cleanup_Loop".
    await client.startNew("Periodic_Cleanup_Loop", instanceId, null);

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

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

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

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

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

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

從永久協調流程離開

如果協調器函式最終一定要完成，則您完全「不」需要呼叫 ContinueAsNew，讓函式自然結束即可。

如果協調器函式處於無限迴圈中，且需要加以停止，請使用協調流程用戶端繫結的終止 API 來將其停止。如需詳細資訊，請參閱執行個體管理。

下一步

了解如何實作單次協調流程

分享方式：