長期協調流程

發行項
04/13/2024

Durable Functions 是 Azure Functions 的擴充功能。您可以使用 協調器函式 來協調函式應用程式中其他 Durable 函式的執行。 Orchestrator 函式具有下列特性：

協調器函式會使用程式代碼定義函式工作流程。不需要宣告式結構描述或設計工具。
協調器函式可以同步和異步地呼叫其他耐久函式。所呼叫函式的輸出可以可靠地儲存至局部變數。
Orchestrator 函式是持久且可靠的。當函式「awaits」或「yields」時，會自動檢查執行進度。當進程回收或 VM 重新啟動時，永遠不會遺失本機狀態。
協調器函式可以長時間執行。協調流程實例的總生命週期可以是秒、天、月或永不結束。

本文提供協調器函式的概觀，以及如何協助您解決各種應用程式開發挑戰。如果您還不熟悉 Durable Functions 應用程式中可用的函式類型，請先閱讀 Durable 函式類型一文。

協調流程身分識別

協調流程的每個執行個體都有執行個體識別碼 (又稱為執行個體識別碼)。根據預設，每個實例標識碼都是自動產生的 GUID。不過，實例標識碼也可以是任何用戶產生的字串值。每個協調流程實例標識碼在工作中樞內都必須是唯一的。

以下是關於實例識別碼的一些規則：

實例標識碼必須介於 1 到 100 個字元之間。
實例標識碼不得以 @開頭。
實例標識碼不得包含 /、 \、 #或 ? 字元。
實例標識碼不得包含控制字元。

注意

通常建議盡可能使用自動產生的執行個體識別碼。用戶產生的實例標識碼適用於協調流程實例與某些外部應用程式特定實體之間有一對一對應的情況，例如採購單或檔。

此外，實際強制執行字元限制規則可能會因應用程式所使用的記憶體提供者而有所不同。為了確保正確的行為和相容性，強烈建議您遵循先前所列的實例標識符規則。

協調流程的實例標識碼是大部分實例管理作業的必要參數。它們對於診斷也很重要，例如在 Application Insights 中搜尋協調流程追蹤數據以進行疑難解答或分析。基於這個理由，建議將產生的實例標識碼儲存到某些外部位置（例如，資料庫或應用程式記錄檔中），以便稍後輕鬆參考。

可靠性

協調器函式會使用事件來源設計模式可靠地維護其執行狀態。長期工作架構會使用僅附加存放區來記錄函式協調流程採用的完整系列動作，而不是直接儲存協調流程的目前狀態。與「傾印」完整運行時間狀態相比，僅限附加存放區有許多優點。優點包括提升效能、延展性和回應性。您也會取得事務數據的最終一致性，以及完整的稽核線索和歷程記錄。稽核線索支援可靠的補償動作。

Durable Functions 會以透明方式使用事件來源。在幕後， await 協調器函式中的（C#）或 yield （JavaScript/Python）運算符會將協調器線程的控制權傳回長期工作架構發送器。在 Java 的情況下，沒有特殊的語言關鍵詞。相反地，呼叫 .await() 工作會透過自定義 Throwable將控制權傳回給發送器。然後發送器會將協調器函式排程的任何新動作（例如呼叫一或多個子函式或排程長期定時器）認可到記憶體。透明認可動作會藉由將所有新事件附加至記憶體，以更新協調流程實例的執行歷程記錄，這與僅限附加的記錄類似。同樣地，認可動作會在記憶體中建立訊息，以排程實際工作。此時，協調器函式可以從記憶體卸除。根據預設，Durable Functions 會使用 Azure 儲存體作為運行時間狀態存放區，但也支援其他記憶體提供者。

當協調流程函式有更多工作要做時（例如收到回應消息或長期定時器到期），協調器會從頭開始喚醒並重新執行整個函式，以重建本機狀態。在重新執行期間，如果程式代碼嘗試呼叫函式（或執行任何其他異步工作），長期工作架構會參考目前協調流程的執行歷程記錄。如果發現活動函式已經執行併產生結果，則會重新執行該函式的結果，且協調器程式代碼會繼續執行。重新執行會繼續執行，直到函式程式碼完成或排程新的異步工作為止。

注意

為了讓重新執行模式正常且可靠地運作，協調器函式程式代碼必須具決定性。不具決定性的協調器程式代碼可能會導致運行時間錯誤或其他非預期的行為。如需協調器函式程式代碼限制的詳細資訊，請參閱 Orchestrator 函式程式代碼條件約束檔。

注意

如果協調器函式發出記錄訊息，重新執行行為可能會導致發出重複的記錄訊息。請參閱記錄主題，以深入瞭解此行為發生的原因，以及如何解決此問題。

協調流程記錄

長期工作架構的事件來源行為與您所撰寫的協調器函式程式代碼緊密結合。假設您有活動鏈結協調器函式，例如下列協調器函式：

注意

適用於 Azure Functions 的第 4 版 Node.js 程式設計模型已正式推出。新的 v4 模型旨在為 JavaScript 和 TypeScript 開發人員提供更靈活的直覺式體驗。如需深入了解 v3 與 v4 之間的差異，請參閱移轉指南。

在下列程式碼片段中，JavaScript (PM4) 表示程式設計模型 V4，這是新的體驗。

[FunctionName("HelloCities")]
public static async Task<List<string>> Run(
    [OrchestrationTrigger] IDurableOrchestrationContext context)
{
    var outputs = new List<string>();

    outputs.Add(await context.CallActivityAsync<string>("SayHello", "Tokyo"));
    outputs.Add(await context.CallActivityAsync<string>("SayHello", "Seattle"));
    outputs.Add(await context.CallActivityAsync<string>("SayHello", "London"));

    // returns ["Hello Tokyo!", "Hello Seattle!", "Hello London!"]
    return outputs;
}

[Function("HelloCities")]
public static async Task<List<string>> Run(
    [OrchestrationTrigger] TaskOrchestrationContext context)
{
    var outputs = new List<string>();

    outputs.Add(await context.CallActivityAsync<string>("SayHello", "Tokyo"));
    outputs.Add(await context.CallActivityAsync<string>("SayHello", "Seattle"));
    outputs.Add(await context.CallActivityAsync<string>("SayHello", "London"));

    // returns ["Hello Tokyo!", "Hello Seattle!", "Hello London!"]
    return outputs;
}

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

module.exports = df.orchestrator(function*(context) {
    const output = [];
    output.push(yield context.df.callActivity("SayHello", "Tokyo"));
    output.push(yield context.df.callActivity("SayHello", "Seattle"));
    output.push(yield context.df.callActivity("SayHello", "London"));

    // returns ["Hello Tokyo!", "Hello Seattle!", "Hello London!"]
    return output;
});

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

df.app.orchestration("helloSequence", function* (context) {
    const output = [];
    output.push(yield context.df.callActivity(helloActivityName, "Tokyo"));
    output.push(yield context.df.callActivity(helloActivityName, "Seattle"));
    output.push(yield context.df.callActivity(helloActivityName, "Cairo"));

    // returns ["Hello Tokyo!", "Hello Seattle!", "Hello Cairo!"]
    return output;
});

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

def orchestrator_function(context: df.DurableOrchestrationContext):
    result1 = yield context.call_activity('SayHello', "Tokyo")
    result2 = yield context.call_activity('SayHello', "Seattle")
    result3 = yield context.call_activity('SayHello', "London")
    return [result1, result2, result3]

main = df.Orchestrator.create(orchestrator_function)

param($Context)

$output = @()

$output += Invoke-DurableActivity -FunctionName 'SayHello' -Input 'Tokyo'
$output += Invoke-DurableActivity -FunctionName 'SayHello' -Input 'Seattle'
$output += Invoke-DurableActivity -FunctionName 'SayHello' -Input 'London'

$output

@FunctionName("HelloCities")
public String helloCitiesOrchestrator(
        @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx) {
    String result = "";
    result += ctx.callActivity("SayHello", "Tokyo", String.class).await() + ", ";
    result += ctx.callActivity("SayHello", "Seattle", String.class).await() + ", ";
    result += ctx.callActivity("SayHello", "London", String.class).await();
    return result;
}

每當排程活動函式時，長期工作架構會將函式的執行狀態檢查點放入一些長期記憶體後端（預設為 Azure 數據表記憶體）。此狀態稱為 協調流程歷程記錄。

記錄資料表

一般而言，長期工作架構會在每個檢查點執行下列動作：

將執行歷程記錄儲存到永久性記憶體。
協調器想要叫用之函式的佇列訊息。
將協調器本身的訊息加入佇列，例如，長期定時器訊息。

檢查點完成後，協調器函式就可以從記憶體中移除，直到有更多工作要做為止。

注意

Azure 儲存體不會在將數據儲存到數據表記憶體和佇列之間提供任何交易保證。為了處理失敗，Durable Functions Azure 儲存體提供者會使用最終一致性模式。這些模式可確保在檢查點中間發生當機或連線中斷時，不會遺失任何數據。替代儲存提供者，例如 Durable Functions MSSQL 記憶體提供者，可能會提供更強大的一致性保證。

完成時，稍早顯示的函式歷程記錄看起來類似 Azure 數據表中的下表儲存體（縮寫為圖例目的）：

PartitionKey （InstanceId）	EventType	時間戳記	輸入	名稱	結果	狀態
eaee885b	ExecutionStarted	2021-05-05T18：45：28.852Z	null	HelloCities
eaee885b	OrchestratorStarted	2021-05-05T18：45：32.362Z
eaee885b	TaskScheduled	2021-05-05T18：45：32.670Z		SayHello
eaee885b	OrchestratorCompleted	2021-05-05T18：45：32.670Z
eaee885b	TaskCompleted	2021-05-05T18：45：34.201Z			“”“Hello Tokyo！””
eaee885b	OrchestratorStarted	2021-05-05T18：45：34.232Z
eaee885b	TaskScheduled	2021-05-05T18：45：34.435Z		SayHello
eaee885b	OrchestratorCompleted	2021-05-05T18：45：34.435Z
eaee885b	TaskCompleted	2021-05-05T18：45：34.763Z			“”“Hello Seattle！””
eaee885b	OrchestratorStarted	2021-05-05T18：45：34.857Z
eaee885b	TaskScheduled	2021-05-05T18：45：34.857Z		SayHello
eaee885b	OrchestratorCompleted	2021-05-05T18：45：34.857Z
eaee885b	TaskCompleted	2021-05-05T18：45：34.919Z			“”“Hello London！””
eaee885b	OrchestratorStarted	2021-05-05T18：45：35.032Z
eaee885b	OrchestratorCompleted	2021-05-05T18：45：35.044Z
eaee885b	ExecutionCompleted	2021-05-05T18：45：35.044Z			“[”“Hello Tokyo！”“，”“Hello Seattle！”“，”“Hello London！”“]”	已完成

資料行值的一些注意事項：

PartitionKey：包含協調流程的實例標識碼。
EventType：代表事件的型別。您可以在這裡找到所有歷程記錄事件類型的詳細描述。
時間戳：記錄事件的UTC時間戳。
名稱：叫用的函式名稱。
輸入：函式的 JSON 格式輸入。
結果：函式的輸出;也就是其傳回值。

警告

雖然它作為偵錯工具很有用，但不要對此數據表有任何相依性。當 Durable Functions 擴充功能進化時，此相依性可能會改變。

每次在等候工作完成之後繼續函式時，長期工作架構會從頭重新執行協調器函式。在每個重新執行時，它會參考執行歷程記錄，以判斷目前的異步工作是否已完成。如果執行歷程記錄顯示工作已完成，架構會重新執行該工作的輸出，並移至下一個工作。此程式會繼續執行，直到重新執行整個執行歷程記錄為止。重新執行目前的執行歷程記錄之後，局部變數將會還原到其先前的值。

功能和模式

下一節說明協調器函式的功能和模式。

子協調流程

協調器函式可以呼叫活動函式，但也呼叫其他協調器函式。例如，您可以從協調器函式的連結庫建置較大的協調流程。或者，您可以平行執行協調器函式的多個實例。

如需詳細資訊和範例，請參閱子協調流程一文。

長期定時器

協調流程可以排程 長期定時器 來實作延遲，或設定異步動作的逾時處理。在協調器函式中使用長期定時器，而不是語言原生的「睡眠」API。

如需詳細資訊和範例，請參閱長期定時器一文。

外部事件

協調器函式可以等候外部事件更新協調流程實例。此 Durable Functions 功能通常用於處理人為互動或其他外部回呼。

如需詳細資訊和範例，請參閱外部事件一文。

錯誤處理

協調器函式可以使用程式設計語言的錯誤處理功能。協調流程程式碼支援 try/catch 之類的現有模式。

協調器函式也可以將重試原則新增至其呼叫的活動或子協調器函式。如果活動或子協調器函式因例外狀況而失敗，則指定的重試原則可以自動延遲並重試執行最多指定的次數。

注意

如果協調器函式中有未處理的例外狀況，協調流程實例將會處於 Failed 狀態。協調流程實例在失敗后無法重試。

如需詳細資訊和範例，請參閱錯誤處理一文。

重要區段（Durable Functions 2.x，目前僅限 .NET）

協調流程實例是單個線程，因此不需要擔心協調流程內的競爭狀況。不過，當協調流程與外部系統互動時，可能會有競爭條件。若要在與外部系統互動時降低競爭狀況，協調器函式可以使用 .NET 中的方法來定義重要區段LockAsync。

下列範例程式代碼顯示定義重要區段的協調器函式。它會使用 LockAsync 方法進入重要區段。此方法需要傳遞一或多個 Durable Entity 的參考，以永久管理鎖定狀態。此協調流程的單一實例一次只能執行重要區段中的程序代碼。

[FunctionName("Synchronize")]
public static async Task Synchronize(
    [OrchestrationTrigger] IDurableOrchestrationContext context)
{
    var lockId = new EntityId("LockEntity", "MyLockIdentifier");
    using (await context.LockAsync(lockId))
    {
        // critical section - only one orchestration can enter at a time
    }
}

LockAsync會取得永久性鎖定，並傳IDisposable回在處置時結束重要區段的。這個 IDisposable 結果可以與區塊搭配 using 使用，以取得重要區段的語法表示法。當協調器函式進入重要區段時，只有一個實例可以執行該程式代碼區塊。嘗試進入重要區段的任何其他實例都會遭到封鎖，直到上一個實例結束重要區段為止。

重要區段功能也有助於協調長期實體的變更。如需重要章節的詳細資訊，請參閱長期實體「實體協調」主題。

注意

Durable Functions 2.0 提供重要區段。目前，只有 .NET 內部協調流程會實作這項功能。 dotnet 隔離背景工作角色的 Durable Functions 中尚未提供實體和重要區段。

呼叫 HTTP 端點（Durable Functions 2.x）

協調器函式不允許執行 I/O，如協調器函式程式碼條件約束中所述。這項限制的典型因應措施是包裝任何需要在活動函式中執行 I/O 的程序代碼。與外部系統互動的協調流程經常使用活動函式進行 HTTP 呼叫，並將結果傳回協調流程。

為了簡化這種常見的模式，協調器函式可以使用方法來直接叫用 CallHttpAsync HTTP API。

[FunctionName("CheckSiteAvailable")]
public static async Task CheckSiteAvailable(
    [OrchestrationTrigger] IDurableOrchestrationContext context)
{
    Uri url = context.GetInput<Uri>();

    // Makes an HTTP GET request to the specified endpoint
    DurableHttpResponse response = 
        await context.CallHttpAsync(HttpMethod.Get, url);

    if ((int)response.StatusCode == 400)
    {
        // handling of error codes goes here
    }
}

為了簡化這種常見的模式，協調器函式可以使用方法來直接叫用 CallHttpAsync HTTP API。針對 C# （隔離式），這項功能是在 Microsoft.Azure.Functions.Worker.Extensions.DurableTask v1.1.0 中引進。

[Function("CheckSiteAvailable")]
public static async Task CheckSiteAvailable(
    [OrchestrationTrigger] TaskOrchestrationContext context)
{
    Uri url = context.GetInput<Uri>();

    // Makes an HTTP GET request to the specified endpoint
    DurableHttpResponse response =
        await context.CallHttpAsync(HttpMethod.Get, url);

    if ((int)response.StatusCode == 400)
    {
        // handling of error codes goes here
    }
}

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

module.exports = df.orchestrator(function*(context) {
    const url = context.df.getInput();
    var res = yield context.df.callHttp("GET", url);
    if (res.statusCode >= 400) {
        // handling of error codes goes here
    }
});

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

df.app.orchestration("checkSiteAvailable", function* (context) {
    const url = context.df.getInput();
    var res = yield context.df.callHttp({ method: "GET", url });
    if (res.statusCode >= 400) {
        // handling of error codes goes here
    }
});

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

def orchestrator_function(context: df.DurableOrchestrationContext):
    url = context.get_input()
    res = yield context.call_http('GET', url)
    if res.status_code >= 400:
        # handing of error code goes here

除了支援基本要求/回應模式之外，此方法還支援自動處理通用異步 HTTP 202 輪詢模式，也支援使用受控識別向外部服務進行驗證。

如需詳細資訊和詳細範例，請參閱 HTTP 功能一文。

注意

從協調器函式直接呼叫 HTTP 端點可在 Durable Functions 2.0 和更新版本中使用。

傳遞多個參數

您無法將多個參數直接傳遞至活動函式。建議傳入物件或複合物件的陣列。

在 .NET 中，您也可以使用 ValueTuple 物件。下列範例使用 C# 7 新增的 ValueTuple 新功能：

[FunctionName("GetCourseRecommendations")]
public static async Task<object> RunOrchestrator(
    [OrchestrationTrigger] IDurableOrchestrationContext context)
{
    string major = "ComputerScience";
    int universityYear = context.GetInput<int>();

    object courseRecommendations = await context.CallActivityAsync<object>(
        "CourseRecommendations",
        (major, universityYear));
    return courseRecommendations;
}

[FunctionName("CourseRecommendations")]
public static async Task<object> Mapper([ActivityTrigger] IDurableActivityContext inputs)
{
    // parse input for student's major and year in university
    (string Major, int UniversityYear) studentInfo = inputs.GetInput<(string, int)>();

    // retrieve and return course recommendations by major and university year
    return new
    {
        major = studentInfo.Major,
        universityYear = studentInfo.UniversityYear,
        recommendedCourses = new []
        {
            "Introduction to .NET Programming",
            "Introduction to Linux",
            "Becoming an Entrepreneur"
        }
    };
}

在 .NET 中，您也可以使用 ValueTuple 物件。下列範例使用 C# 7 新增的 ValueTuple 新功能：

[Function("GetCourseRecommendations")]
public static async Task<object> RunOrchestrator(
    [OrchestrationTrigger] TaskOrchestrationContext context, int universityYear)
{
    string major = "ComputerScience";

    object courseRecommendations = await context.CallActivityAsync<object>(
        "CourseRecommendations",
        (major, universityYear));
    return courseRecommendations;
}

[FunctionName("CourseRecommendations")]
public static async Task<object> Mapper(
    [ActivityTrigger] (string Major, int UniversityYear) studentInfo, FunctionContext executionContext)
{
    // retrieve and return course recommendations by major and university year
    return new
    {
        major = studentInfo.Major,
        universityYear = studentInfo.UniversityYear,
        recommendedCourses = new []
        {
            "Introduction to .NET Programming",
            "Introduction to Linux",
            "Becoming an Entrepreneur"
        }
    };
}

協調器

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

module.exports = df.orchestrator(function*(context) {
    const location = {
        city: "Seattle",
        state: "WA"
    };
    const weather = yield context.df.callActivity("GetWeather", location);

    // ...
});

`GetWeather` 活動

module.exports = async function (context, location) {
    const {city, state} = location; // destructure properties into variables

    // ...
};

const getWeatherActivityName = "getWeather";

df.app.orchestration("getWeatherOrchestrator", function* (context) {
    const location = {
        city: "Seattle",
        state: "WA",
    };
    const weather = yield context.df.callActivity(getWeatherActivityName, location);

    // ...
});

df.app.activity(getWeatherActivityName, async function (location) {
    const { city, state } = location; // destructure properties into variables

    // ...
});

協調器

from collections import namedtuple
import azure.functions as func
import azure.durable_functions as df

def orchestrator_function(context: df.DurableOrchestrationContext):
    Location = namedtuple('Location', ['city', 'state'])
    location = Location(city='Seattle', state= 'WA')

    weather = yield context.call_activity("GetWeather", location)

    # ...

`GetWeather` 活動

from collections import namedtuple

Location = namedtuple('Location', ['city', 'state'])

def main(location: Location) -> str:
    city, state = location
    return f"Hello {city}, {state}!"

協調器

param($Context)

$output = @()

$location = @{
    City = 'Seattle'
    State  = 'WA'
}

Invoke-ActivityFunction -FunctionName 'GetWeather' -Input $location

# ...

`GetWeather` 活動

param($location)

"Hello $($location.City), $($location.State)!"
# ...

@FunctionName("GetWeatherOrchestrator")
public String getWeatherOrchestrator(
        @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx) {
    var location = new Location();
    location.city = "Seattle";
    location.state = "WA";
    String weather = ctx.callActivity("GetWeather", location, String.class).await();
    return weather;
}

@FunctionName("GetWeather")
public String getWeather(@DurableActivityTrigger(name = "location") Location location) {
    if (location.city.equals("Seattle") && location.state.equals("WA")) {
        return "Cloudy";
    } else {
        return "Unknown";
    }
}

class Location {
    public String city;
    public String state;
}

下一步

Orchestrator 程式代碼條件約束

長期協調流程

協調流程身分識別

可靠性

協調流程記錄

記錄資料表

功能和模式

子協調流程

長期定時器

外部事件

錯誤處理

重要區段 （Durable Functions 2.x，目前僅限 .NET）

呼叫 HTTP 端點 （Durable Functions 2.x）

傳遞多個參數

下一步

其他資源

重要區段（Durable Functions 2.x，目前僅限 .NET）

呼叫 HTTP 端點（Durable Functions 2.x）