c# - 具有链接和并行工作流程的 Azure Durable 功能

Question

在我的项目中，大约有 6 个 azure 函数需要根据某些 JSON 配置按顺序运行。

配置如下

{
  "configurations": [
    {
      "Name": "Function1",
      "Sequence": 1
    },
    {
      "Name": "Function2",
      "Sequence": 2
    },
    {
      "Name": "Function3",
      "Sequence": 2
    },
    {
      "Name": "Function4",
      "Sequence": 3
    },
    {
      "Name": "Function5",
      "Sequence": 4
    },
    {
      "Name": "Function6",
      "Sequence": 5
    }
  ]
}

这意味着，它应该首先触发 function1，然后是 function2 和 function3。当 function2 和 3 都完成时，函数 4 应该执行..然后 5,6

我知道编排函数中的链接是这样的

    [FunctionName(nameof(FunctionChaining))]
    public static async Task<List<string>> FunctionChainingRunOrchestrator(
        [OrchestrationTrigger] 
        IDurableOrchestrationContext context)
    {
        // Serial calls
        await context.CallActivityAsync<string>(
            nameof(Function1), "param1");            
        await context.CallActivityAsync<string>(
            nameof(Function2), "param2");
        await context.CallActivityAsync<string>(
            nameof(Function3), "param3");

    }

以及并行调用

    [FunctionName(nameof(parallelcalls))]
    public static async Task<List<string>> ParallelOrchestrator(
        [OrchestrationTrigger]
        IDurableOrchestrationContext context)
    {
        // Parallel calls
        var tasks = new List<Task<string>>();
        tasks.Add(context.CallActivityAsync<string>(
            nameof(Function2),
            "param1"));
        tasks.Add(context.CallActivityAsync<string>(
            nameof(Function3),
            "param2"));
        await Task.WhenAll(tasks);
    }

所以我需要触发函数1，然后当它在await Task.WhenAll(tasks);时一起完成2和3时，我需要触发函数4和5作为链接

但我真的不知道如何根据我的配置使其成为工作流程。我的耐用功能只是一天...

请帮忙

Answer 1

您可以动态创建您需要调用的任务。

按顺序对配置的任务进行分组，当为同一序列配置多个任务时，您可以并行运行这些任务。

class Configuration {
    public string Name { get;set; }
    public int Sequence { get;set; }
}

[FunctionName(nameof(Run))]
public static async Task<List<string>> RunConfiguredTasks(
    [OrchestrationTrigger] IDurableOrchestrationContext context
)
{
    List<Configuration> configuredTasks = ...; // however you get the configuration

    // iterate over all task groups, grouped by sequence number
    foreach (var configGroup in configuredTasks.GroupBy(t => t.Sequence))
    {
        var tasks = new List<Task<string>>();

        // iterate over all tasks in the task group
        foreach (var config in configGroup)
        {
            var taskInput = ... // get input for the task
            tasks.Add(context.CallActivityAsync<string>(config.Name, taskInput));
        }

        // wait for all tasks in the current task group
        await Task.WhenAll(tasks);
    }
}

(未经测试的代码，但应该很接近:p)