如何为Durable Functions并行活动实现实时进度上报

完全可以实现需求下的N/5粒度进度反馈，不需要额外搭建独立的状态上报链路，直接依托Durable Functions自带的自定义状态能力就能完成。

实现思路 #

你目前使用的set_custom_status接口本身支持在编排执行的任意阶段动态更新状态，完全可以承载并行批次的细分进度。
只需要调整现有“一次性等待所有并行任务完成后再更新状态”的逻辑：

• 拿到待处理的条目列表后，先初始化进度统计结构，记录总任务数、已完成数、每个条目的独立运行状态
• 给每个并行触发的活动任务绑定完成回调，每当任意一个活动执行结束，就实时更新进度统计值
• 每一次进度更新都会直接写入编排实例的自定义状态，用户发起实例状态查询时就能直接拿到最新的N/5统计结果，甚至每个条目的单独运行状态。

调整后的可运行示例代码 #

# 第一阶段：拉取待处理条目
context.set_custom_status({
    "stage": "fetching items",
    "total_tasks": 0,
    "completed_tasks": 0,
    "item_details": []
})
items = yield context.call_activity('do_thing', None)  # 示例返回列表 [1, 2, 3, 4, 5]

# 初始化批次进度状态
total_item_count = len(items)
batch_progress = {
    "stage": "processing batch items",
    "total_tasks": total_item_count,
    "completed_tasks": 0,
    "item_details": [
        {"item_id": item, "status": "running"}
        for item in items
    ]
}
context.set_custom_status(batch_progress)

# 定义单任务完成后的进度更新逻辑
def update_progress_on_item_done(item_id):
    # 匹配对应条目，更新为完成状态
    for detail in batch_progress["item_details"]:
        if detail["item_id"] == item_id:
            detail["status"] = "completed"
    batch_progress["completed_tasks"] += 1
    # 实时把最新进度写入自定义状态
    context.set_custom_status(batch_progress)

# 构建并行任务列表，绑定进度更新回调
batch_tasks = []
for item in items:
    parallel_task = context.call_activity('do_another_thing', item)
    # 任务执行完成后触发进度更新
    parallel_task.then(lambda _, current_item=item: update_progress_on_item_done(current_item))
    batch_tasks.append(parallel_task)

# 等待所有并行任务执行完成
results = yield context.task_all(batch_tasks)

# 全部任务完成后更新最终状态
batch_progress["stage"] = "complete"
context.set_custom_status(batch_progress)

落地注意要点 #

• 所有状态更新操作必须放在编排器的确定执行流程内，不要在活动函数内部直接修改编排器的自定义状态，避免因为Durable Functions的编排重放机制导致状态值错乱
• 匹配条目状态时要以任务绑定的静态item_id为依据，不要直接依赖任务返回的动态值做匹配，保证重放时的状态一致性
• 前端对接查询接口时，直接提取实例返回结构中的customStatus字段做展示即可，不需要额外处理其他状态数据

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