5.2 - Activity 执行
开启 Activity 执行
如何使用 Temporal Python SDK 启动 Activity 执行。
启动 Activity 执行的调用写在 Workflow 定义内部。启动 Activity 执行的调用将生成 ScheduleActivityTask 命令。这将在工作流执行事件历史中产生与 Activity 任务相关的三个事件:ActivityTaskScheduled、ActivityTaskStarted 和 ActivityTaskClosed。
多个并发 Activity 调用之间共享 Activity 实现的单个实例。Activity 实现代码应具备幂等性。
通过调用参数传递给 Activity 或通过返回值返回的值都将被记录到执行历史中。当工作流状态需要恢复时,整个执行历史将从 Temporal 服务传输到工作流 Worker。因此,过大的执行历史可能对工作流的性能产生不利影响。
因此,务必注意通过 Activity 调用参数或返回值传输的数据量。除此之外,Activity 实现没有其他额外限制。
要启动 Activity 执行,需要在工作流定义中使用 execute_activity() 操作。
execute_activity() 是 start_activity() 的快捷方式,前者在后者的结果上等待。
如果只想获取句柄,以便单独进行等待或取消操作,可使用 start_activity()。在大多数情况下,除非需要高级任务能力,否则使用 execute_activity()。
传递给 Activity 的单个参数是位置参数。start_activity() 或 execute_activity() 的类型安全形式不支持多个参数,如果传递多个参数,必须通过 args 关键字参数提供。
from datetime import timedelta
from temporalio import workflow
with workflow.unsafe.imports_passed_through():
from your_activities_dacx import your_activity
from your_dataobject_dacx import YourParams
@workflow.defn(name="YourWorkflow")
class YourWorkflow:
@workflow.run
async def run(self, name: str) -> str:
return await workflow.execute_activity(
your_activity,
YourParams("Hello", name),
start_to_close_timeout=timedelta(seconds=10),
)设置必需的 Activity 超时时间
如何使用 Temporal Python SDK 设置必需的 Activity 超时时间。
Activity 执行语义依赖多个参数。唯一必须设置的值是 Schedule-To-Close Timeout 或 Start-To-Close Timeout。这些值在 Activity 选项(Activity Option)中进行设置。
Activity 选项是在 Activity 参数之后传入的关键字参数。
可用的超时时间包括:
schedule_to_close_timeout
schedule_to_start_timeout
start_to_close_timeout
with workflow.unsafe.imports_passed_through():
from activities import your_activity, YourParams
@workflow.defn
class YourWorkflow:
@workflow.run
async def run(self, greeting: str) -> list[str]:
activity_timeout_result =await workflow.execute_activity(
your_activity,
YourParams(greeting,"Activity Timeout option"),
# Activity Execution Timeout
start_to_close_timeout=timedelta(seconds=10),
# schedule_to_start_timeout=timedelta(seconds=10),
# schedule_to_close_timeout=timedelta(seconds=10),
)
# ...获取 Activity 执行的结果
如何使用 Temporal Python SDK 获取 Activity 执行的结果。
启动 Activity 执行的调用将生成 ScheduleActivityTask 命令,并且为工作流提供 Awaitable 对象。工作流执行可以选择阻塞等待,直到通过 Awaitable 获取到结果,也可以继续推进执行,待结果可用时再使用。
使用 start_activity() 可以启动 Activity,返回其句柄 ActivityHandle。使用 execute_activity() 可以直接返回执行结果。
必须提供 schedule_to_close_timeout 或 start_to_close_timeout 中的至少一个。
execute_activity() 是 await start_activity() 的快捷方式。该辅助函数与 start_activity() 接受相同的参数,同时等待返回结果。在大多数情况下,除非需要高级任务能力,否则应使用 execute_activity()。
from datetime import timedelta
from temporalio import workflow
with workflow.unsafe.imports_passed_through():
from your_activities_dacx import your_activity
from your_dataobject_dacx import YourParams
@workflow.defn(name="YourWorkflow")
class YourWorkflow:
@workflow.run
async def run(self, name: str) -> str:
return await workflow.execute_activity(
your_activity,
YourParams("Hello", name),
start_to_close_timeout=timedelta(seconds=10),
)