4.2 - 子工作流
启动子工作流执行(Child Workflow Execution)
子工作流执行是指从另一个工作流内部通过子工作流 API 进行调度的工作流执行。
使用子工作流 API 时,与子工作流相关的事件(比如 StartChildWorkflowExecutionInitiated、ChildWorkflowExecutionStarted、ChildWorkflowExecutionCompleted)将被记录到工作流执行的事件历史中。
ChildWorkflowExecutionStarted 事件必须在父工作流完成之前记录到事件历史中,以确保子工作流已经启动。在 Python 中,await start_child_workflow() 或 execute_child_workflow() 将在内部等待此事件记录完成后才返回,因此一旦调用完成,即可保证子工作流已经启动。如果从非主协程(比如信号或更新处理器)中启动子工作流,则需确保父工作流在该调用完成之前不会结束。
要在 Python 中生成子工作流执行,可以使用 execute_child_workflow() 函数,该函数启动子工作流,并且等待其完成;也可以使用 start_child_workflow() 函数启动子工作流,该函数返回子工作流执行的句柄。如果希望在子工作流启动后继续执行操作,或者想获取工作流/Run ID,或者想在子工作流运行期间向其发送信号,那么使用后者。
execute_child_workflow() 辅助函数等同于 start_child_workflow() 加上 await handle。
from temporalio import workflow
from dataobject import ComposeGreetingInput
from temporalio.workflow import ParentClosePolicy
@workflow.defn
class ComposeGreetingWorkflow:
@workflow.run
async def run(self, input: ComposeGreetingInput) -> str:
return f"{input.greeting}, {input.name}!"
@workflow.defn
class GreetingWorkflow:
@workflow.run
async def run(self, name: str) -> str:
return await workflow.execute_child_workflow(
ComposeGreetingWorkflow.run,
ComposeGreetingInput("Hello", name),
id="hello-child-workflow-workflow-child-id",
parent_close_policy=ParentClosePolicy.ABANDON,
)设置父关闭策略
父关闭策略决定当父工作流变为关闭状态(已完成、失败或超时)时,子工作流执行将受到何种影响。
默认的父关闭策略选项被设置为终止子工作流执行。
在 start_child_workflow() 或 execute_child_workflow() 函数中设置 parent_close_policy 参数,指定父工作流关闭时子工作流的行为。
from temporalio.workflow import ParentClosePolicy
# ...
@workflow.defn
class ComposeGreetingWorkflow:
@workflow.run
async def run(self, input: ComposeGreetingInput) -> str:
return f"{input.greeting}, {input.name}!"
@workflow.defn
class GreetingWorkflow:
@workflow.run
async def run(self, name: str) -> str:
return await workflow.execute_child_workflow(
ComposeGreetingWorkflow.run,
ComposeGreetingInput("Hello", name),
id="hello-child-workflow-workflow-child-id",
parent_close_policy=ParentClosePolicy.ABANDON,
)