如何在Azure Durable Functions扇出扇入模式中完成所有任务（无视失败）

你遇到的问题是task_all会在任一任务失败时立即抛出异常并停止等待其余任务，要实现无论任务成功或失败都等待全部完成，可以通过给每个任务添加异常捕获包装的方式解决，具体实现如下：

核心思路 #

给每个活动任务调用套一层异常捕获逻辑，让失败的任务返回错误信息而非直接抛出异常，这样task_all就会等待所有包装后的任务执行完毕，不会中途终止。

代码示例 #

import azure.durable_functions as df
from azure.durable_functions.models import DurableOrchestrationContext

async def orchestrator_function(context: DurableOrchestrationContext):
    # 定义需要执行的原始任务列表
    original_tasks = [
        context.call_activity("ProcessData", "data1"),
        context.call_activity("ProcessData", "data2"),
        context.call_activity("ProcessData", "data3")
    ]

    # 包装任务：捕获异常，统一返回格式
    async def wrap_task(task):
        try:
            result = await task
            return {"status": "success", "output": result}
        except Exception as err:
            return {"status": "failed", "error_msg": str(err)}

    # 生成包装后的任务列表
    wrapped_tasks = [wrap_task(task) for task in original_tasks]

    # 等待所有包装任务完成
    all_results = await context.task_all(wrapped_tasks)

    # 后续可以按需处理结果，比如统计成功/失败数量、提取结果等
    success_count = sum(1 for res in all_results if res["status"] == "success")
    fail_count = len(all_results) - success_count

    return {
        "total_tasks": len(all_results),
        "success_count": success_count,
        "fail_count": fail_count,
        "details": all_results
    }

方案说明 #

1. 包装任务：通过wrap_task函数捕获每个任务的异常，将成功结果和失败错误统一封装为字典格式，避免异常直接抛出终止task_all的等待。
2. 等待全部任务：task_all等待所有包装后的任务执行完成，无论原始任务成功或失败，都会得到对应的结果条目。
3. 结果处理：最后可以根据返回的结果字典，灵活统计或提取成功/失败的任务信息。

这种方式完全适配Azure Durable Functions Python的任务模型，替代了JS版本中Promise.allSettled的实现逻辑，确保所有任务都会执行完毕。

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