跳到主要内容

高级：高可用背压

当目标父 Flow 可能因过载拒绝请求时，使用短生命周期的 SubmitRequestFlow。

Definition graph

SubmitRequestFlow

Valid

python · examples/python/dex_examples/patterns/parallel-subflows/submit_request_flow.py

客户端启动 SubmitRequestFlow，而不是在内存中重试 RPC。SubmitStep 对请求分区，调用目标父 Flow 的 SendRequest RPC，并且仅在父 Flow 接受请求后完成。false 响应会让 Step 失败，普通 Step 重试策略随后执行 durable backoff 并再次尝试。父 Flow 的启动竞争也由同一套重试处理：FlowAlreadyStarted 会让 SubmitStep 失败，下一次执行再在已活跃的父 Flow 上调用 RPC。即使提交进程或 Worker 重启，Dex 也会保留重试状态。

这会分离接入和处理：父 Flow 保持有界，提交请求则保持 durable，并可独立重试。

核心实现​

class SubmitStep(Step[SubmitRequestInput]):
    def __init__(
        self,
        client_provider: Callable[[], AsyncClient],
        parent_flow: AdvancedShortLiveParentFlow,
    ) -> None:
        self.client_provider = client_provider
        self.parent_flow = parent_flow

    async def execute(  # type: ignore[override]
        self, context: AsyncContext, input: SubmitRequestInput
    ) -> StepDecision:
        if not input.parent_ids:
            raise ValueError("at least one parent Flow ID is required")
        parent_id = input.parent_ids[partition(input.request, len(input.parent_ids))]
        accepted = await enqueue_request(
            self.client_provider(), self.parent_flow, parent_id, input.request
        )
        if not accepted:
            raise RuntimeError(f"parent {parent_id} rejected the request")
        return graceful_complete(parent_id)


async def enqueue_request(
    client: AsyncClient,
    parent_flow: AdvancedShortLiveParentFlow,
    parent_id: str,
    request: str,
) -> bool:
    try:
        return await client.invoke_rpc(parent_flow.send_request, parent_id, request)
    except FlowNotActiveError:
        await client.start_flow(
            parent_flow,
            parent_id,
            ParentInput([request], DEFAULT_CONCURRENCY),
            StartFlowOptions(id_reuse_policy=IdReusePolicy.ALLOW_IF_NOT_RUNNING),
        )
        return True


class SubmitRequestFlow(Flow[SubmitRequestInput]):
    def __init__(
        self,
        client_provider: Callable[[], AsyncClient],
        parent_flow: AdvancedShortLiveParentFlow,
    ) -> None:
        self.submit = SubmitStep(client_provider, parent_flow)

    def get_steps(self) -> StepList[SubmitRequestInput]:
        return StepList.start_step(self.submit)

例子: examples/python/dex_examples/patterns/parallel-subflows/submit_request_flow.py

相关内容​