Azure Durable Function编排器竞争条件问题及解决方法咨询

你的问题是典型的分布式竞争条件，属于使用预设实例ID的Azure Durable Functions场景下的常见问题。当两条消息几乎同时到达时，第一条消息触发的start_new操作尚未完成实例初始化，第二条消息的状态检查会返回实例未运行的结果，导致它也尝试启动新实例。虽然Durable Functions运行时最终会阻止重复创建相同ID的实例，但过程中会出现实例覆盖或异常，破坏消息收集逻辑。

核心原因 #

is_function_running状态检查与start_new实例创建操作并非原子性操作，两者之间存在时间间隙。在这个间隙内，另一个请求可能已经发起了实例创建请求，导致前置状态检查的结果失效，进而引发竞争。

解决方法 #

移除前置的状态检查逻辑，直接尝试启动编排实例，利用Durable Functions运行时的原子性实例创建机制规避竞争。当捕获到"实例已存在"的异常时，直接发送外部事件即可。这种方式能彻底消除竞争窗口，因为start_new操作在运行时层面是互斥的，不会同时创建两个相同ID的实例。

此外，确保编排器中的计时器逻辑正确：每次收到新消息后重置等待计时器，保证最后一条消息发送后能等待完整的超时时间再执行合并操作。

调整后的队列触发器代码 #

import json
import asyncio

import azure.functions as func

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

from lib.logtrail import get_logger
logtrail = get_logger("on_new_message_fragment")


async def main(message: func.QueueMessage, starter: str):
    orchestrator = df.DurableOrchestrationClient(starter)

    payload = json.loads(message.get_body().decode("utf-8"))
    user_phone = payload["phone"]
    instance_id = f"message-collector:{user_phone}"

    try:
        # 直接尝试启动实例，跳过前置状态检查
        logtrail.info(f"Attempting to start or send event to {instance_id} with {payload}")
        await orchestrator.start_new(
            "collect_messages",
            instance_id=instance_id,
            client_input=payload
        )
    except Exception as e:
        if "already exists" in str(e).lower():
            # 实例已存在或正在创建，发送外部事件追加消息
            logtrail.info(f"Instance {instance_id} already exists; appending message via event")
            await orchestrator.raise_event(instance_id, "new-message", payload)
        else:
            logtrail.error(f"Failed to process message: {e}")
            raise

编排器代码优化（增强健壮性） #

原有编排器逻辑已正确实现"收到新消息重置计时器"的功能，这里补充事件数据解析的健壮性处理：

import os
import json

import azure.durable_functions as df
from datetime import timedelta
from lib.logtrail import get_logger

logtrail = get_logger("message_collector")

def collect_messages(context: df.DurableOrchestrationContext):
    user_message = context.get_input()
    timeout_duration = int(os.getenv("MESSAGE_COLLECTOR_TIMEOUT_DURATION", 10))
    full_message = {**user_message}

    while True:
        message_task = context.wait_for_external_event("new-message")
        timeout_task = context.create_timer(context.current_utc_datetime + timedelta(seconds=timeout_duration))

        completed_task = yield context.task_any([message_task, timeout_task])

        if completed_task == timeout_task:
            timeout_task.cancel()
            break

        # 健壮性处理：兼容不同格式的事件数据
        try:
            new_message = completed_task.result
            if isinstance(new_message, str):
                new_message = json.loads(new_message)
        except json.JSONDecodeError as e:
            logtrail.error(f"Failed to parse new message event: {e}")
            timeout_task.cancel()
            continue

        full_message = yield context.call_activity("append_message_fragment", {
            "full_message": full_message,
            "message_fragment": new_message
        })

    yield context.call_activity("send_message_for_analysis", full_message)

main = df.Orchestrator.create(collect_messages)

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