Azure 持久编排功能触发两次

我正在尝试实现 Azure Durable Function 工作流程。

每隔 6 分钟，我就会有一个 Azure TimerTrigger 函数调用一个 Azure Orchestration Function (OrchestrationTrigger)，该函数又会启动多个活动函数 (ActivityTrigger)。

然而，有时，Orchestration 函数会在几秒钟内被调用两次！这是一个大问题，因为我的活动函数不是幂等的！

下面是我的代码是如何被调用的。

定时器触发功能：

[FunctionName("StartupFunc")]
public static async Task Run([TimerTrigger("0 */6 * * * *", RunOnStartup = true, UseMonitor = false)]TimerInfo myStartTimer, [OrchestrationClient] DurableOrchestrationClient orchestrationClient, TraceWriter log)
{
    List<OrchestrationModel> ExportModels = await getData();

    string id = await orchestrationClient.StartNewAsync("OrchestratorFunc", ExportModels);
}

编排功能：

[FunctionName("OrchestratorFunc")]
public static async Task<string> TransformOrchestration([OrchestrationTrigger] DurableOrchestrationContext context, TraceWriter log)
{
    var dataList = context.GetInput<List<OrchestrationModel>>();
    var tasks = new List<Task>();

    foreach (var data in dataList)
    {
        tasks.Add(context.CallActivityAsync<string>("TransformToSql", new TransformModel(data));
    }
    await Task.WhenAll(tasks);
}

活动功能：

[FunctionName("TransformToSql")]
[public static async Task<string> RunTransformation([ActivityTrigger] DurableActivityContext context, TraceWriter log)
{
    TransformModel = context.GetInput<TransformModel>();

    //Do some work with TransformModel
}

这种行为完全没问题——这就是 Durable Functions 的设计原理。

您有以下编排：

[FunctionName("OrchestratorFunc")]
public static async Task<string> TransformOrchestration([OrchestrationTrigger] DurableOrchestrationContext context, TraceWriter log)
{
    var dataList = context.GetInput<List<OrchestrationModel>>();
    var tasks = new List<Task>();

    foreach (var data in dataList)
    {
        tasks.Add(context.CallActivityAsync<string>("TransformToSql", new TransformModel(data));
    }
    await Task.WhenAll(tasks);
}

当一个活动被调用时，流程返回到一个概念，称为调度员- 它是 Durable Functions 的内部存在，负责维持编排流程。当它等待任务完成时，编排会暂时释放。一旦任务完成，整个编排就会重播，直到下一个任务await发生。

重要的是，尽管重播了编排，但活动不会再次被调用 - 它的结果是从存储中获取并使用的。为了让事情更明确，请考虑以下示例：

[FunctionName("OrchestratorFunc")]
public static async Task<string> TransformOrchestration([OrchestrationTrigger] DurableOrchestrationContext context, TraceWriter log)
{
    var dataList = context.GetInput<List<OrchestrationModel>>();
    var tasks = new List<Task>();

    foreach (var data in dataList)
    {
        await context.CallActivityAsync<string>("TransformToSql1", new TransformModel(data);
        await context.CallActivityAsync<string>("TransformToSql2", new TransformModel(data);
    }
}

When TransformToSql1等待，编排被释放，整个流程等待该活动完成。然后重播编曲——它再次等待TransformToSql1但由于它的结果已保存，它只是返回到编排并等待TransformToSql2- 然后重复该过程。