标签: azure-durable-functions

我想按需处理数百万条记录，大约需要 2-3 个小时来处理。我想要无服务器，这就是为什么尝试持久功能（第一次）。我想检查一下，我可以运行持久函数多长时间，所以我创建了 3 个函数

1. 用于启动 Orchestrator 功能的 Http 功能
2. 编排器功能
3. 活动功能

我的 DurableFunction 正在运行并在 Application Insights 中发出过去 5 天的日志，根据我的代码，它需要 15 天才能完成。

我想知道如何手动停止 Orchestrator 功能？

我可以在 ApplicationInsights 请求表中看到数千个单次执行条目，有没有办法检查后端运行了多少个 DurableFunction？以及单次执行需要多少时间？

我可以在“DurableFunctionHubInstance”表中看到一些有关协调器功能的信息，但正如 MS 建议的那样，不要依赖表。

我有一个具有持久函数（以及一些非持久函数）的函数应用程序，并且我收到一个 InvalidOperationException ，其中显示：Unable to find an Azure Storage connection string to use for this binding.

这是堆栈跟踪：

System.InvalidOperationException:
  at Microsoft.Azure.WebJobs.Extensions.DurableTask.DurableTaskExtension.GetOrchestrationServiceSettings (Microsoft.Azure.WebJobs.Extensions.DurableTask, Version=1.5.0.0, Culture=neutral, PublicKeyToken=null)
  at Microsoft.Azure.WebJobs.Extensions.DurableTask.DurableTaskExtension.Microsoft.Azure.WebJobs.Host.Config.IExtensionConfigProvider.Initialize (Microsoft.Azure.WebJobs.Extensions.DurableTask, Version=1.5.0.0, Culture=neutral, PublicKeyToken=null)
  at Microsoft.Azure.WebJobs.Host.Executors.JobHostConfigurationExtensions.InvokeExtensionConfigProviders (Microsoft.Azure.WebJobs.Host, Version=2.3.0.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35)
  at Microsoft.Azure.WebJobs.Host.Executors.JobHostConfigurationExtensions.CreateStaticServices (Microsoft.Azure.WebJobs.Host, Version=2.3.0.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35)
  at Microsoft.Azure.WebJobs.JobHost.InitializeServices (Microsoft.Azure.WebJobs.Host, Version=2.3.0.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35)
  at Microsoft.Azure.WebJobs.Script.Utility.CreateMetadataProvider (Microsoft.Azure.WebJobs.Script, Version=1.0.0.0, Culture=neutral, PublicKeyToken=nullMicrosoft.Azure.WebJobs.Script, Version=1.0.0.0, Culture=neutral, PublicKeyToken=null: C:\projects\azure-webjobs-sdk-script\src\WebJobs.Script\Utility.csMicrosoft.Azure.WebJobs.Script, Version=1.0.0.0, Culture=neutral, PublicKeyToken=null: 362)
  at Microsoft.Azure.WebJobs.Script.ScriptHost.LoadBindingExtensions (Microsoft.Azure.WebJobs.Script, Version=1.0.0.0, Culture=neutral, PublicKeyToken=nullMicrosoft.Azure.WebJobs.Script, Version=1.0.0.0, Culture=neutral, PublicKeyToken=null: C:\projects\azure-webjobs-sdk-script\src\WebJobs.Script\Host\ScriptHost.csMicrosoft.Azure.WebJobs.Script, Version=1.0.0.0, Culture=neutral, PublicKeyToken=null: 966)
  at Microsoft.Azure.WebJobs.Script.ScriptHost.Initialize (Microsoft.Azure.WebJobs.Script, Version=1.0.0.0, …

我正在查看此示例以在设置超时后运行持久函数 Activity。

https://docs.microsoft.com/en-us/azure/azure-functions/durable/durable-functions-eternal-orchestrations

这将允许我的函数活动执行数据处理，然后在它尝试再次加载之前等待 1 小时。这将永远持续下去。完美的。

但是，将函数发布到 Azure 时，我不想通过关联的 HTTP 触发器手动调用/启动函数。我只希望持久功能自动启动并开始处理。

这可能吗？如果没有，建议的解决方法是什么？

谢谢！

我有一个持久函数，每天由计时器触发器触发一次：

[FunctionName("MyDurableFunction")]
public static async Task Run(
    [TimerTrigger("0 0 23 * * *", RunOnStartup = false)] TimerInfo myTimer,
    [OrchestrationClient] DurableOrchestrationClient starter,
    ILogger log)
{
    await starter.StartNewAsync("OrchestrationFunction", null);
}

[FunctionName("OrchestrationFunction")]
public static async Task OrchestrationFunction(
    [OrchestrationTrigger]DurableOrchestrationContext context,
    ILogger log)
{
    // do stuff
}

这很好用。出于测试目的，我还希望能够通过 Http 触发器触发持久功能，因此我添加了以下内容：

[FunctionName("MyDurableFunctionHttpTrigger")]
public static async Task<IActionResult> Run(
    [HttpTrigger(AuthorizationLevel.Anonymous, "get", "post", Route = "demo")]HttpRequest req, 
    [OrchestrationClient] DurableOrchestrationClient starter, 
    ILogger log)
{
    await starter.StartNewAsync("OrchestrationFunction", null);
    return new OkObjectResult("blah");
}

在本地运行这些触发器（包括 http 触发器或计时器触发器）将触发该函数，但将两者都包含在类中意味着这两个触发事件都不会发生。是否可以让多个触发器类型启动编排触发器？

我有一个耐用的功能扇出和输入模式，但似乎工作不可靠。每 10 分钟从 Timer 函数调用一次 Orchestration，但此后已将其增加到 20 分钟。使用 context.CallActivityAsync 调用 Activity 函数并返回一个整数（已处理的行数）。目前，传入的 workItems 应该只包含 2 个要处理的项目。第一项处理所有行并在日志中显示完整的行。第二项有时显示正在处理的行，但在某些时候它只是停止...没有其他活动被识别，并且“完成”永远不会显示在日志中。另外，第二个活动有时会显示它同时运行多次...我已经在我的开发机器上使用相同的数据尝试了这个确切的代码，并且它处理完成的时间不超过 5 分钟。我还将hosts.json 文件设置为

{
  "version": "2.0",
  "functionTimeout": "00:10:00",
  "extensions": {
    "queues": {
      "maxPollingInterval": "00:00:05"
    }
  }
}

编排：

public static async void RunOrchestrator(
            [OrchestrationTrigger] DurableOrchestrationContext context, ILogger log)
        {
            log.LogInformation($"************** Fanning out ********************");
            var parallelTasks = new List<Task<int>>();
            //object[] workBatch = await context.CallActivityAsync<object[]>("GETVendors", null);
            object[] workBatch = GETVendors(log); //work batch only has 2 items 
            for (int i = 0; i<workBatch.Length; i++)
            {
                Task<int> task …

我发布这个问题是为了看看我是否正确理解 Azure Functions 中的并行性，尤其是 Durable Functions。

最近使用 az cli 将设置最大并行度的功能添加到 Azure Functions：https : //github.com/Azure/azure-functions-host/issues/1207

az resource update --resource-type Microsoft.Web/sites -g <resource_group> -n <function_app_name>/config/web --set properties.functionAppScaleLimit=<scale_limit>

我已将此应用于我的函数应用程序，但我不确定它如何与持久函数的 MaxConcurrentOrchestratorFunctions 和 MaxConcurrentActivityFunctions 设置一起使用。

以下是否会导致全局最多 250 个并发活动功能？

• 功能应用规模限制：5
• MaxConcurrentOrchestratorFunctions：5
• MaxConcurrentActivityFunctions：10

作为 Azure Durable Functions 领域的新手，我正在努力寻找处理对已实施速率限制的 API 的下游调用的最佳方法。

我的流程设置方式如下：

• HistorySynchronizer_HttpStart函数：在签名中具有HttpTrigger和绑定，并调用下一个编排函数：DurableClient
• HistorySynchronizer功能：OrchestrationTrigger在签名中有绑定。该函数将调用 API（等待）并返回集合。对于该集合中的每个项目，它将启动一个新的 Activity：（context.CallActivityAsync()将它们组合在 a 中List并执行 a Task.WhenAll()）
• ProcessActivity函数：ActivityTrigger在签名中具有绑定。该函数必须调用速率受限的 API 端点。我想要限制的正是这些活动（跨多个编排）。

所以，我正在寻找的是节流模式的实现：

• 我需要一个线程安全的共享状态（我正在考虑持久实体）来跟踪对该 API 的调用次数
• 在 ProcessActivity 函数调用该 API 之前，它必须检查持久实体是否可以调用该 API，或者是否必须在执行该调用之前等待一定的 TimeSpan。
• 如果必须等待，活动就必须“睡眠”，我正在考虑持久计时器，但似乎应该在编排功能中使用，而不是在活动功能中使用。
• 如果允许调用 API，活动必须更新该共享状态对象中的速率计数器。

我没有看到实现此目的的标准实现，并且我希望将尽可能多的检查/睡眠逻辑从主编排中移出。

最好的方法是为每个必须限制的 API 调用实现子编排，在调用 Activity 之前必须进行检查吗？

期待任何见解。

我有一个持久函数，其输入由先前的活动函数确定。对于每个活动函数，我有多个可等待的任务，其中每个任务都依赖于前一个任务的输出。

这是我的结构如下：

协调者

[FunctionName("MessageController")]
public static async void Run(
    [OrchestrationTrigger] DurableOrchestrationContext context,
    TraceWriter log)
{
    if (!context.IsReplaying) log.Warning("MessageController started");

    var Input1= context.CallActivityAsync<ResultMessage>("Function_1", new InputMessage());
    var Input2= context.CallActivityAsync<ResultMessage>("Function_2", Input1);
    var Input3= context.CallActivityAsync<ResultMessage>("Function_2", Input2);

}

活动功能

[FunctionName("Function_1")]
public static ResultMessage Run(
    [ActivityTrigger] DurableActivityContext activityContext,
    TraceWriter log)
{
    //Awaitable task
    var taskOutput= await DoSomething();

    //Awaitable task
    var token = await DoAnotherThing(taskOutput);
}

我已经对此进行了测试，一切正常。但我想知道这是否是好的做法？对于持久函数来说，在活动函数中存在可等待任务是否正常？

计划在 Azure 上根据消费定价模型配置基于“队列触发”的 Azure 功能。每次调用此函数时，可能会随机运行 1 分钟到 30 分钟之间的任何时间，甚至更长/更短，具体取决于情况。偶尔最大。运行时间可能约为 45 分钟，否则大部分是最大运行时间。运行时间可以在10分钟以内。

基于“队列触发”的 Azure 功能是否存在超时？如果是，最大值是多少？暂停？

是否有任何基于 Azure 功能相关的实践来克服超时限制？

（或者）

“耐用功能”可以被视为克服最大值的替代方案吗？超时时间限制？

（或者）

Azure 上还有其他更好的无服务器组件可供使用吗？

有几个相互调用的持久函数。 主编排 -> 子编排 -> Activity -> Helper 异步方法

每个函数都有 ILogger 依赖性，并在函数开始和函数结束时登录。由于某种原因，两个协调器都会重复“启动时”消息。（见图）活动没有这种效果。（见图）多次运行下面的示例 - 同一个故事。

我也确信整个过程已经被触发过一次。

这是协调器中的错误还是预期行为？

using System;
using System.Net.Http;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Extensions.DurableTask;
using Microsoft.Azure.WebJobs.Extensions.Http;
using Microsoft.Extensions.Logging;

namespace Issues
{
    public static class Log_Issue
    {
        [FunctionName("Main")]
        public static async Task RunOrchestrator(
            [OrchestrationTrigger] IDurableOrchestrationContext context,
            ILogger log)
        {
            try
            {
                log.LogWarning("Main Start");
                await context.CallSubOrchestratorAsync("Sub", null);
                log.LogWarning("Main End");
            }
            catch (Exception e)
            {
                Console.WriteLine(e);
                throw;
            }
        }

        [FunctionName("Sub")]
        public static async Task RunSubOrchestrator(
            [OrchestrationTrigger] IDurableOrchestrationContext context,
            ILogger …