你当前正在访问 Microsoft Azure Global Edition 技术文档网站。如果需要访问由世纪互联运营的 Microsoft Azure 中国技术文档网站，请访问 https://docs.azure.cn。

持久业务流程

项目
04/29/2024

Durable Functions 是 Azure Functions 的一个扩展。可以使用业务流程协调程序函数来协调函数应用中其他持久函数的执行。业务流程协调程序函数具有以下特征：

业务流程协调程序函数使用过程代码来定义函数工作流。不需要声明性的架构或设计器。
业务流程协调程序函数可以同步和异步调用其他持久函数。被调用函数的输出可以可靠保存到本地变量。
业务流程协调程序函数是持久且可靠的。当函数处于“等待”或“生成”状态时，可自动对执行进度设置检查点。回收进程或重启 VM 时，从来不会丢失本地状态。
业务流程协调程序函数可以长时间运行。业务流程实例的总生存期可以是几秒、几天、几月或者永不结束。

本文概述业务流程协调程序功能，及其如何帮助解决各种应用开发难题。如果你不熟悉 Durable Functions 应用中可用的函数类型，请先阅读持久函数类型一文。

业务流程标识

业务流程的每个实例都有一个实例标识符（也称为“实例 ID”）。默认情况下，每个实例 ID 都是自动生成的 GUID。但是，实例 ID 也可以是用户生成的任何字符串值。每个业务流程实例 ID 在任务中心内必须唯一。

下面是有关实例 ID 的一些规则：

实例 ID 必须为 1 到 100 个字符。
实例 ID 不能以 @ 开头。
实例 ID 不能包含 /、\、# 或 ? 字符。
实例 ID 不能包含控制字符。

注意

通常建议尽可能使用自动生成的实例 ID。用户生成的实例 ID 适用于业务流程实例与某些外部应用程序特定的实体（例如采购订单或文档）之间存在一对一映射的情况。

此外，字符限制规则的实际强制实施可能会有所不同，具体取决于应用使用的存储提供程序。为了确保正确的行为和兼容性，强烈建议遵循前面列出的实例 ID 规则。

业务流程的实例 ID 是大多数实例管理操作的必需参数。它们对于诊断也很重要，例如，在 Application Insights 中搜索业务流程跟踪数据以进行故障排除或分析。出于此原因，建议将生成的实例 ID 保存到可在今后方便引用的某个外部位置（例如数据库或应用程序日志中）。

可靠性

业务流程协调程序函数使用事件溯源设计模式可靠地维护自身的执行状态。 Durable Task Framework 使用仅限追加的存储来记录函数业务流程执行的一系列完整操作，而不是直接存储业务流程的当前状态。与“转储”整个运行时状态相比，仅限追加的存储具有很多优点。优点包括提高性能、可伸缩性和响应能力。此外，它还提供事务数据的最终一致性，以及完整的审核线索和历史记录。审核线索支持可靠的补偿操作。

Durable Functions 以透明方式使用事件溯源。在后台，业务流程协调程序函数中的 await (C#) 或 yield (JavaScript/Python) 运算符将对业务流程协调程序线程的控制权让回给 Durable Task Framework 调度程序。对于 Java，没有特殊的语言关键字。相反，在某个任务上调用 .await() 将通过自定义 Throwable 把控制权交还给调度程序。然后，该调度程序向存储提交业务流程协调程序函数计划的任何新操作（如调用一个或多个子函数或计划持久计时器）。透明提交操作通过将所有新事件追加到存储中来更新业务流程实例的执行历史记录，就像只追加日志一样。同样，提交操作在存储中创建消息以计划实际工作。此时，可从内存中卸载业务流程协调程序函数。默认情况下，Durable Functions 使用 Azure 存储作为其运行时状态存储，但还支持其他存储提供程序。

如果业务流程函数需要执行其他工作（例如，收到响应消息或持久计时器到期），业务流程协调程序将会唤醒，并从头开始重新执行整个函数，以重新生成本地状态。在重播过程中，如果代码尝试调用函数（或执行任何其他异步工作），Durable Task Framework 会查询当前业务流程的执行历史记录。如果该扩展发现活动函数已执行并已生成结果，则会回放该函数的结果并且业务流程协调程序代码继续运行。在函数代码完成或计划了新的异步工作之前，重放会一直继续。

注意

要使重播模式正常可靠工作，业务流程协调程序函数代码必须是确定性的。非确定性业务流程协调程序代码可能会导致运行时错误或其他意外行为。有关业务流程协调程序函数的代码限制的详细信息，请参阅业务流程协调程序函数代码约束文档。

注意

如果业务流程协调程序函数发出日志消息，重播行为可能导致发出重复的日志消息。请参阅日志记录主题，详细了解此行为发生的原因及其解决方法。

业务流程历史记录

Durable Task Framework 的事件溯源行为与编写的业务流程协调程序函数代码密切相关。假设你有一个活动链接业务流程协调程序函数，如以下业务流程协调程序函数：

注意

适用于 Azure Functions 的 Node.js 编程模型版本 4 现已正式发布。新版 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;
}

每当计划活动函数时，Durable Task Framework 会在某些持久存储后端（默认为 Azure 表存储）中对该函数的执行状态触发检查点操作。此状态称为“业务流程历史记录”。

历史记录表

一般而言，Durable Task Framework 会在每个检查点位置执行以下操作：

将执行历史记录保存到持久存储中。
将业务流程协调程序想要调用的函数的相关消息排队。
将业务流程协调程序本身的消息排队例如，持久计时器消息。

完成检查点之后，在业务流程协调程序函数需要执行其他工作之前，可以从内存中任意删除该函数。

注意

在将数据保存到表存储和队列的间隔期限内，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.67Z		SayHello
eaee885b	OrchestratorCompleted	2021-05-05T18:45:32.67Z
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：包含业务流程的实例 ID。
EventType：表示事件的类型。可在此处找到所有历史记录事件类型的详细说明。
时间戳：历史记录事件的 UTC 时间戳。
名称：调用的函数的名称。
输入：函数的 JSON 格式的输入。
Result：函数的输出，即其返回值。

警告

尽管此表可以用作有效的调试工具，但不要对它有任何依赖。它可能会随着 Durable Functions 扩展的演变而变化。

每当函数在等待任务完成后恢复时，Durable Task Framework 都会从头开始重新运行业务流程协调程序函数。每次重新运行时，它都会查询执行历史记录，确定当前的异步任务是否已完成。如果执行历史记录显示任务已完成，Framework 会重播此任务的输出，并转到下一个任务。此过程会持续到整个历史记录被重播为止。重播当前执行历史记录后，局部变量已还原到其先前值。

功能和模式

以下部分介绍业务流程协调程序函数的功能和模式。

子业务流程

业务流程协调程序函数可以调用活动函数，但也可以调用其他业务流程协调程序函数。例如，可以基于业务流程协调程序函数库构建更大的业务流程。或者，你可以并行运行某个业务流程协调程序函数的多个实例。

有关详细信息和示例，请参阅子业务流程一文。

持久计时器

业务流程可以计划持久计时器来实现延迟或设置处理异步操作时的超时。请在业务流程协调程序函数中使用持久计时器，而不要使用语言原生的“sleep”API。

有关详细信息和示例，请参阅持久计时器一文。

外部事件

业务流程协调程序函数可以等待外部事件来更新业务流程实例。此项 Durable Functions 功能通常用于处理人机交互或其他外部回调。

有关详细信息和示例，请参阅外部事件一文。

错误处理。

业务流程协调程序函数可以使用编程语言的错误处理功能。业务流程代码中支持诸如 try/catch 的现有模式。

业务流程协调程序函数还可以将重试策略添加到它们调用的活动或子业务流程协调程序函数。如果某个活动或子业务流程协调程序函数失败并出现异常，则指定的重试策略可以自动延迟并重试执行，直到达到指定的次数。

注意

如果业务流程协调程序函数中出现未经处理的异常，则业务流程实例将以 Failed 状态完成。业务流程实例在失败后不可重试。

有关详细信息和示例，请参阅错误处理一文。

关键节（Durable Functions 2.x，当前仅限 .NET）

业务流程实例是单线程的，因此无需考虑业务流程内部的争用情况。但是，当业务流程与外部系统交互时，可能会出现争用情况。若要在与外部系统交互时缓解争用情况，业务流程协调程序函数可以使用 .NET 中的 LockAsync 方法定义关键节。

以下示例代码演示了一个定义关键节的业务流程协调程序函数。它使用 LockAsync 方法进入关键节。此方法要求向某个持久管理锁状态的持久实体传递一个或多个引用。此业务流程的单个实例每次只能执行关键节中的代码。

[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 功能一文。

注意

在 Durable Functions 2.0 及更高版本中可以直接从业务流程协调程序函数调用 HTTP 终结点。

传递多个参数

无法直接将多个参数传递给一个活动函数。建议传入对象或复合对象的数组。

在 .NET 中，还可使用 ValueTuples 对象。以下示例使用了 C# 7 添加的 ValueTuples 的新功能：

[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 中，还可使用 ValueTuples 对象。以下示例使用了 C# 7 添加的 ValueTuples 的新功能：

[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;
}

后续步骤

业务流程协调程序代码约束

通过

持久业务流程

业务流程标识

可靠性

业务流程历史记录

历史记录表

功能和模式

子业务流程

持久计时器

外部事件

错误处理。

关键节（Durable Functions 2.x，当前仅限 .NET）

调用 HTTP 终结点 (Durable Functions 2.x)

传递多个参数

业务流程协调程序

`GetWeather` 活动

业务流程协调程序

`GetWeather` 活动

业务流程协调程序

`GetWeather` 活动

后续步骤

反馈

其他资源