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 参数之后传入的关键字参数。

可用的超时时间包括：

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),
        )