5.3 - Activity 超时

设置 Activity 超时时间

每类 Activity 超时时间分别控制 Activity 执行中不同方面的最大持续时间。

Activity 选项（Activity Option）中提供以下超时时间：

Activity 执行必须设置 Start-To-Close Timeout 或 Schedule-To-Close Timeout 中的至少一个。

Activity 选项是在 Activity 参数后面传入的关键字参数。

可用的超时时间包括：

from datetime import timedelta

from temporalio import workflow
from temporalio.common import RetryPolicy

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

设置 Activity 重试策略

重试策略与超时时间配合使用，提供精细化控制手段，优化执行体验。

如果未提供自定义重试策略，Activity 执行自动关联默认的重试策略。

要在 Python 中创建 Activity 重试策略，需要在 start_activity() 或 execute_activity() 函数中设置 RetryPolicy 类。

from datetime import timedelta

from temporalio import workflow
from temporalio.common import RetryPolicy

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_result =await workflow.execute_activity(
            your_activity,
            YourParams(greeting,"Retry Policy options"),
            start_to_close_timeout=timedelta(seconds=10),
# Retry Policy
            retry_policy=RetryPolicy(
                backoff_coefficient=2.0,
                maximum_attempts=5,
                initial_interval=timedelta(seconds=1),
                maximum_interval=timedelta(seconds=2),
# non_retryable_error_types=["ValueError"],
            ),
        )
        return activity_result

使用 next_retry_delay 重写重试间隔

要在 Activity 中覆盖当前策略设置的下一次重试间隔，需要在抛出 ApplicationError 时传入 next_retry_delay 参数。该值将替换并覆盖重试策略中原本计算出的重试间隔。

比如可以根据 Activity 的尝试次数，设置延迟间隔。在下面的示例中，第一次尝试失败后的重试延迟从 3 秒开始，第二次尝试增加到 6 秒，第三次增加到 9 秒，以此类推。这种方式创建稳步递增的退避，与退避系数使用指数方式有所不同：

from temporalio.exceptions import ApplicationError
from datetime import timedelta

@activity.defn
async def my_activity(input: MyActivityInput):
    try:
        # Your activity logic goes here
    except Exception as e:
        attempt = activity.info().attempt
        raise ApplicationError(
            f"Error encountered on attempt {attempt}",
            next_retry_delay=timedelta(seconds=3 * attempt),
        ) from e

Activity 心跳

Activity 心跳是执行 Activity 的 Worker 进程向 Temporal 服务端发送的“心跳”信号。心跳告诉 Temporal 服务端：该 Activity 执行正在推进，Worker 尚未崩溃。如果在心跳超时时间段内，Temporal 服务端未收到心跳，则将该 Activity 视为失败，可能根据重试策略重新调度另一次 Activity 任务执行。

心跳并非总是被发送到 Temporal 服务端——Worker 可能对心跳进行限流。

Activity 取消请求在 Activity 发送心跳时由 Temporal 服务端传递给 Activity。不发送心跳的 Activity 无法接收取消请求。心跳限流可能导致取消请求的送达晚于预期。

心跳可以包含 details 字段，用于描述 Activity 当前的执行进度。如果重试 Activity，该 Activity 可以访问上次发送给 Temporal 服务端的心跳中的详细信息。

要在 Python 中向 Activity 执行发送心跳，需使用 heartbeat() API。

@activity.defn
async def your_activity_definition() -> str:
    activity.heartbeat("heartbeat details!")

除获取取消信息外，心跳还支持携带详细数据，这些数据被持久化存储在服务端，以便在 Activity 重试时进行检索。如果 Activity 调用 heartbeat(123, 456)，随后失败，并且重试，则在下一次运行时，heartbeat_details 将返回包含 123 和 456 的可迭代对象。

设置心跳超时

心跳超时与 Activity 心跳配合使用。

heartbeat_timeout 是 start_activity() 函数的参数，用于设置 Activity 心跳之间的最大时间间隔。

workflow.start_activity(
    activity="your-activity",
    schedule_to_close_timeout=timedelta(seconds=5),
    heartbeat_timeout=timedelta(seconds=1),
)

execute_activity() 是 start_activity() 的快捷方式，前者等待后者的结果上等待。

如果只想获取句柄，以便单独进行等待或取消操作，需使用 start_activity()。在大多数情况下，除非需要高级任务能力，否则应使用 execute_activity()。

workflow.execute_activity(
    activity="your-activity",
    name,
    schedule_to_close_timeout=timedelta(seconds=5),
    heartbeat_timeout=timedelta(seconds=1),
)