Azure编排器中Activity Function重复执行的原因与解决方法

问题描述 #

在Azure环境运行以下Durable Orchestrator函数时，发现某一Activity Function的日志被多次记录。请分析该现象产生的原因，并说明如何确保除扇出函数外的Activity Function仅执行一次。

函数代码 #

[FunctionName("ExportOrchestratorFunction")]
public async Task RunOrchestrator([OrchestrationTrigger] IDurableOrchestrationContext context)
{
    var processId = context.InstanceId;
    var custNo = _configuration.GetValue<string>("CustNo");
    

    var roads = await context.CallActivityAsync<IEnumerable<RoadReport>>("GetCustomersByRoadFunction", new RoadCustomerRetrievalParameters
    {
        CustNo = custNo ?? "*",
        Road = RoadName ?? "*",
        ActiveFlag = "*",
        APIFlag = "*",
        ProcessId = processId
    });

    var roadReports = string.Join(',', roads.Select(r => r.Road).ToArray());
    _logger.LogInformation($"Retrieved roads with customers count: { roadReports }. Process ID: {processId}");

    var tasks = new List<Task<RoadExportData>>();

    // Fan -out
    foreach (var road in roads)
    {
        _logger.LogInformation($"Creating retrieval task for {road.Road}. Process ID: {processId}");
        var task = context.CallActivityAsync<RoadExportData>(
            "RoadRetrievalFunction",
            new RoadRetrievalParameters
            {
                Road = road.Road,
                ProcessId = processId,
                CustNos = road.Customers
            }
        );

        tasks.Add(task);
    }

    _logger.LogInformation($"Starting all tasks to retrieve Data. Process ID: {processId}");
    // Fan -in: Wait for all tasks to complete
    var exportResults = await Task.WhenAll(tasks);

    // Aggregate the results
    var allExportData = exportResults.ToList();
    _logger.LogInformation($"All ETA retrieved { allExportData }");

    await context.CallActivityAsync<int>("LogAndPersistFunction", new PersistDataParameters
    {
        ExportData = allExportData,
        ProcessId = processId
    });
}

原因分析 #

• Orchestrator重放机制导致日志混淆：Durable Orchestrator基于历史记录重放执行逻辑，当主机重启、实例扩展或流程续跑时，会重新执行整个代码流程。如果在Orchestrator中直接使用普通日志记录（比如foreach循环内的日志），重放时这些日志会被重复输出，容易被误认为是Activity函数的日志重复。
• Activity函数自动重试：Durable Task Framework默认对失败的Activity函数执行重试（默认3次），如果Activity因超时、异常等原因执行失败，会被多次调用，导致其内部日志重复记录。
• 扇出输入存在重复项：如果roads集合中包含重复的Road值，会导致多次调用同一个Road对应的RoadRetrievalFunction，进而产生重复日志。

解决方案 #

1. 处理Orchestrator重放日志问题 #

使用Orchestrator的重放安全日志记录方式，避免重放时重复输出日志：

// 方式一：通过IsReplaying判断是否为非重放状态
if (!context.IsReplaying)
{
    _logger.LogInformation($"Creating retrieval task for {road.Road}. Process ID: {processId}");
}

// 方式二：创建重放安全的日志记录器
var replaySafeLogger = context.CreateReplaySafeLogger(_logger);
replaySafeLogger.LogInformation($"Creating retrieval task for {road.Road}. Process ID: {processId}");

2. 控制Activity函数的重试策略 #

根据业务需求配置Activity的重试规则，比如禁用重试或设置合理的重试次数：

// 在Orchestrator调用Activity时配置重试选项
var task = context.CallActivityAsync<RoadExportData>(
    "RoadRetrievalFunction",
    parameters,
    new ActivityOptions
    {
        // 禁用重试
        RetryOptions = new RetryOptions(TimeSpan.FromSeconds(5), 0)
    });

3. 确保扇出输入无重复 #

在扇出前对roads集合去重，避免重复调用同一Activity：

var uniqueRoads = roads
    .GroupBy(r => r.Road)
    .Select(g => g.First())
    .ToList();

foreach (var road in uniqueRoads)
{
    // 原有的任务创建逻辑
}

4. 实现Activity函数的幂等性 #

即使Activity被意外重复调用，也要保证业务结果唯一。可以通过ProcessId + Road作为唯一标识，在Activity内部先校验是否已处理过该请求：

[FunctionName("RoadRetrievalFunction")]
public async Task<RoadExportData> RunActivity(
    [ActivityTrigger] RoadRetrievalParameters parameters,
    ILogger logger)
{
    // 先检查是否已处理过该ProcessId+Road的请求
    bool isProcessed = await CheckIfProcessed(parameters.ProcessId, parameters.Road);
    if (isProcessed)
    {
        logger.LogInformation($"Request {parameters.ProcessId}-{parameters.Road} already processed.");
        // 返回已处理的结果
        return await GetExistingResult(parameters.ProcessId, parameters.Road);
    }

    // 原有的业务逻辑
    // ...

    // 标记请求已处理
    await MarkAsProcessed(parameters.ProcessId, parameters.Road);
    return result;
}

内容的提问来源于stack exchange，提问作者Dilshan Prasad