Durable Functions 中的任务中心 (Azure Functions)

Durable Functions 中的任务中心表示应用程序在存储中的当前状态，包括所有挂起的工作。 当函数应用运行时，业务流程、活动和实体函数的进度将持续存储在任务中心。 这可确保应用程序能在它停止的地方恢复处理（如果它因某种原因暂时停止或中断后需要重启）。 此外，它使函数应用能够动态缩放计算辅助角色。

从概念上讲，任务中心存储以下信息：

• 所有业务流程和实体实例的实例状态。
• 要处理的消息，包括
  • 表示等待运行的活动的任何活动消息。
  • 正在等待传递到实例的任何实例消息。

活动与实例消息的区别在于活动消息是无状态消息，因此可以在任意位置进行处理，而实例消息需要传递到由实例 ID 标识的特定有状态实例（业务流程或实体）。

在内部，每个存储提供程序可能使用不同的组织来表示实例状态和消息。 例如，消息由 Azure 存储提供程序存储在 Azure 存储队列中，但由 MSSQL 提供程序存储在关系表中。 就应用程序的设计而言，这些差异并不重要，但其中一些可能会影响性能特征。 我们将在下面的存储中的表示部分讨论它们。

工作项

任务中心内的活动消息和实例消息表示函数应用需要处理的工作。 函数应用运行时，它会从任务中心持续提取工作项。 每个工作项都在处理一个或多个消息。 我们区分两种类型的工作项：

• 活动工作项：运行活动函数来处理活动消息。
• 业务流程协调程序工作项：运行业务流程协调程序或实体函数来处理一个或多个实例消息。

辅助角色可以同时处理多个工作项，但须遵守配置的每个辅助角色并发限制。

辅助角色完成工作项后，会将效果提交回任务中心。 这些效果因执行的函数类型而异：

• 已完成的活动函数会创建一个包含结果的实例消息，发送给父业务流程协调程序实例。
• 已完成的业务流程协调程序函数会更新业务流程状态和历史，并可能创建新消息。
• 已完成的实体函数会更新实体状态，还可能创建新的实例消息。

对于业务流程，每个工作项代表该业务流程执行的一个关卡。 当有新消息需要业务流程协调程序处理时，一个关卡就开始了。 此类消息可能指示业务流程应启动；可能指示活动、实体调用、计时器或子业务流程已完成；也可能指示外部事件。 该消息触发工作项，使业务流程协调程序能够处理结果并继续下一个关卡。 当业务流程协调程序完成或到达必须等待新消息的点时，该关卡结束。

执行示例

请考虑一个扇出扇入业务流程，该业务流程并行启动两个活动，并等待两个活动完成：

[FunctionName("Example")]
public static async Task Run([OrchestrationTrigger] IDurableOrchestrationContext context)
{
    Task t1 = context.CallActivityAsync<int>("MyActivity", 1);
    Task t2 = context.CallActivityAsync<int>("MyActivity", 2);
    await Task.WhenAll(t1, t2);
}

module.exports = df.orchestrator(function*(context){
    const tasks = [];
    tasks.push(context.df.callActivity("MyActivity", 1));
    tasks.push(context.df.callActivity("MyActivity", 2));
    yield context.df.Task.all(tasks);
});

def orchestrator_function(context: df.DurableOrchestrationContext):
    tasks = []
    tasks.append(context.call_activity("MyActivity", 1))
    tasks.append(context.call_activity("MyActivity", 2))
    yield context.task_all(tasks)

param($Context)

$Tasks = @()

$Tasks += Invoke-DurableActivity -FunctionName 'MyActivity' -Input 1 -NoWait
$Tasks += Invoke-DurableActivity -FunctionName 'MyActivity' -Input 2 -NoWait

Wait-DurableTask -Task $Tasks

@FunctionName("Example")
public void exampleOrchestrator(
        @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx) {
    Task<Void> t1 = ctx.callActivity("MyActivity", 1);
    Task<Void> t2 = ctx.callActivity("MyActivity", 2);
    ctx.allOf(List.of(t1, t2)).await();
}

客户端启动此业务流程后，函数应用将其作为一系列工作项进行处理。 每个已完成的工作项在提交时都会更新任务中心状态。 下面是步骤：

1. 客户端请求启动实例 ID 为“123”的新业务流程。 客户端完成此请求后，任务中心包含业务流程状态的占位符和实例消息：

   

   标签 ExecutionStarted 是众多历史事件类型之一，用于标识参与业务流程历史的各种类型的消息和事件。

2. 辅助角色执行业务流程协调程序工作项来处理 ExecutionStarted 消息。 它调用开始执行业务流程代码的业务流程协调程序函数。 此代码计划两个活动，然后在等待结果时停止执行。 辅助角色提交此工作项后，任务中心包含

   

   运行时状态现在为 Running，添加了两条新的 TaskScheduled 消息，并且历史记录现在包含五个事件，分别为：OrchestratorStarted、ExecutionStarted、TaskScheduled、TaskScheduled、OrchestratorCompleted。 这些事件代表此业务流程执行的第一个关卡。

3. 辅助角色执行活动工作项来处理其中一条 TaskScheduled 消息。 它使用输入“2”调用活动函数。 活动函数完成后，它会创建包含结果的 TaskCompleted 消息。 辅助角色提交此工作项后，任务中心包含

   

4. 辅助角色执行业务流程协调程序工作项来处理 TaskCompleted 消息。 如果业务流程仍缓存在内存中，则只需恢复执行。 否则，辅助角色首先重播历史记录以恢复业务流程的当前状态。 然后，它会继续业务流程，交付活动的结果。 收到此结果后，业务流程仍在等待其他活动的结果，因此它再次停止执行。 辅助角色提交此工作项后，任务中心包含

   

   业务流程历史记录现在包含另外三个事件：OrchestratorStarted、TaskCompleted 和 OrchestratorCompleted。 这些事件代表此业务流程执行的第二个关卡。

5. 辅助角色执行活动工作项来处理剩余的 TaskScheduled 消息。 它使用输入“1”调用活动函数。 辅助角色提交此工作项后，任务中心包含

   

6. 辅助角色执行业务流程协调程序工作项来处理 TaskCompleted 消息。 收到第二个结果后，业务流程将完成。 辅助角色提交此工作项后，任务中心包含

   

   运行时状态现在为 Completed，业务流程历史记录现在包含另外四个事件，分别为：OrchestratorStarted、TaskCompleted、ExecutionCompleted、OrchestratorCompleted。 这些事件代表此业务流程执行的第三个关卡和最后一关卡。

然后，此业务流程执行的最终历史记录包含 12 个事件，分别为：OrchestratorStarted、ExecutionStarted、TaskScheduled、TaskScheduled、OrchestratorCompleted、OrchestratorStarted、TaskCompleted、OrchestratorCompleted、OrchestratorStarted、TaskCompleted、ExecutionCompleted、OrchestratorCompleted。

任务中心管理

接下来，让我们更深入地了解如何创建或删除任务中心、如何在运行多个函数应用时正确使用任务中心，以及如何检查任务中心的内容。

创建和删除

首次启动函数应用时，会自动在存储中创建包含全部所需资源的空任务中心。

如果使用默认的 Azure 存储提供程序，则无需额外配置。 否则，请按照配置存储提供程序的说明确保存储提供程序可以正确预配和访问任务中心所需的存储资源。

多个函数应用

如果多个函数应用共享存储帐户，则必须使用单独的任务中心名称配置每个函数应用。 此要求也适用于过渡槽：每个过渡槽必须配置有唯一的任务中心名称。 单个存储帐户可以包含多个任务中心。 此限制通常也适用于其他存储提供程序。

下图说明了在共享和专用 Azure 存储帐户中每个函数应用的一个任务中心。

内容检查

可通过几种常见方法来检查任务中心的内容：

1. 在函数应用中，客户端对象提供查询实例存储的方法。 若要详细了解支持的查询类型，请参阅实例管理一文。
2. 同样，HTTP API 提供 REST 请求来查询业务流程和实体的状态。 有关更多详细信息，请参阅 HTTP API 参考。
3. Durable Functions 监视器工具可以检查任务中心并提供各种视觉显示选项。

对于某些存储提供程序，还可以直接转到基础存储来检查任务中心：

• 如果使用 Azure 存储提供程序，则实例状态将存储在实例表和历史记录表中，可以使用 Azure 存储资源管理器等工具进行检查。
• 如果使用 MSSQL 存储提供程序，可以使用 SQL 查询和工具检查数据库中的任务中心内容。

存储中的表示形式

每个存储提供程序使用不同的内部组织来表示存储中的任务中心。 在对函数应用进行故障排除或尝试确保性能、可伸缩性或成本目标时，了解此组织（虽然不需要）可能会有所帮助。 因此，我们简要介绍了每个存储提供程序如何在存储中组织数据。 有关各种存储提供程序选项及其比较方式的详细信息，请参阅 Durable Functions 存储提供程序一文。

Azure 存储提供程序

Azure 存储提供程序使用以下组件表示存储中的任务中心：

• 两个 Azure 表存储实例状态。
• 一个 Azure 队列存储活动消息。
• 一个或多个 Azure 队列存储实例消息。 每个所谓的控制队列都表示一个分区，根据实例 ID 的哈希分配所有实例消息的子集。
• 几个额外的 Blob 容器，用于租用 Blob 和/或大型消息。

例如，名为 xyz 的任务中心 (PartitionCount = 4) 包含以下队列和表：

接下来，我们将更详细地介绍这些组件及其扮演的角色。

有关 Azure 存储提供程序如何表示任务中心的详细信息，请参阅 Azure 存储提供程序一文。

Netherite 存储提供程序

Netherite 将所有任务中心状态划分为指定数量的分区。 在存储中，使用以下资源：

• 一个 Azure 存储 blob 容器，其中包含所有 blob 并按分区分组。
• 一个 Azure 表，其中包含有关分区的已发布指标。
• 一个 Azure 事件中心命名空间，用于在分区之间传递消息。

例如，名为 mytaskhub 的任务中心 (PartitionCount = 32) 在存储中表示如下：

Netherite 使用基于日志和检查点的事件溯源机制来表示分区的当前状态。 使用块 blob 和页 blob。 无法直接从存储中读取此格式，因此在查询实例存储时必须运行函数应用。

有关 Netherite 存储提供程序的任务中心的详细信息，请参阅 Netherite 存储提供程序的任务中心信息。

MSSQL 存储提供程序

所有任务中心数据都存储在一个关系数据库中，但使用多个表：

• dt.Instances 和 dt.History 表存储实例状态。
• dt.NewEvents 表存储实例消息。
• dt.NewTasks 表存储活动消息。

为了使多个任务中心能够在同一数据库中独立共存，每个表都包含一个 TaskHub 列作为其主键的一部分。 与其他两个提供程序不同，MSSQL 提供程序没有分区的概念。

有关 MSSQL 存储提供程序的任务中心的详细信息，请参阅 Microsoft SQL (MSSQL) 存储提供程序的任务中心信息。

任务中心名称

任务中心由名称标识，该名称必须符合以下规则：

• 仅包含字母数字字符
• 以字母开头
• 最小长度为 3 个字符，最大长度为 45 个字符

任务中心名称在 host.json 文件中声明，如以下示例所示：

host.json (Functions 2.0)

{
  "version": "2.0",
  "extensions": {
    "durableTask": {
      "hubName": "MyTaskHub"
    }
  }
}

host.json (Functions 1.x)

{
  "durableTask": {
    "hubName": "MyTaskHub"
  }
}

还可以使用应用设置配置任务中心，如以下 host.json 示例文件所示：

host.json (Functions 1.0)

{
  "durableTask": {
    "hubName": "%MyTaskHub%"
  }
}

host.json (Functions 2.0)

{
  "version": "2.0",
  "extensions": {
    "durableTask": {
      "hubName": "%MyTaskHub%"
    }
  }
}

任务中心名称将设置为 MyTaskHub 应用设置的值。 以下 local.settings.json 演示了如何将 MyTaskHub 设置定义为 samplehubname：

{
  "IsEncrypted": false,
  "Values": {
    "MyTaskHub" : "samplehubname"
  }
}

除了 host.json，还可以在业务流程客户端绑定元数据中配置任务中心名称。 如果需要访问位于单独的函数应用中的业务流程或实体，此功能很有用。 以下代码演示如何编写使用业务流程客户端绑定来处理配置为应用设置的任务中心的函数：

[FunctionName("HttpStart")]
public static async Task<HttpResponseMessage> Run(
    [HttpTrigger(AuthorizationLevel.Function, methods: "post", Route = "orchestrators/{functionName}")] HttpRequestMessage req,
    [DurableClient(TaskHub = "%MyTaskHub%")] IDurableOrchestrationClient starter,
    string functionName,
    ILogger log)
{
    // Function input comes from the request content.
    object eventData = await req.Content.ReadAsAsync<object>();
    string instanceId = await starter.StartNewAsync(functionName, eventData);

    log.LogInformation($"Started orchestration with ID = '{instanceId}'.");

    return starter.CreateCheckStatusResponse(req, instanceId);
}

{
    "name": "input",
    "taskHub": "%MyTaskHub%",
    "type": "durableClient",
    "direction": "in"
}

{
    "name": "input",
    "taskHub": "%MyTaskHub%",
    "type": "durableClient",
    "direction": "in"
}

{
    "name": "input",
    "taskHub": "%MyTaskHub%",
    "type": "durableClient",
    "direction": "in"
}

@FunctionName("HttpStart")
public HttpResponseMessage httpStart(
        @HttpTrigger(name = "req", route = "orchestrators/{functionName}") HttpRequestMessage<?> req,
        @DurableClientInput(name = "durableContext", taskHub = "%MyTaskHub%") DurableClientContext durableContext,
        @BindingName("functionName") String functionName,
        final ExecutionContext context) {

    DurableTaskClient client = durableContext.getClient();
    String instanceId = client.scheduleNewOrchestrationInstance(functionName);
    context.getLogger().info("Created new Java orchestration with instance ID = " + instanceId);
    return durableContext.createCheckStatusResponse(req, instanceId);
}

任务中心名称必须以字母开头且只能包含字母和数字。 如果未指定，则会使用默认的任务中心名称，如下表所示：

有关扩展版本之间差异的详细信息，请参阅 Durable Functions 版本一文。

后续步骤