5.1 - Activity 基础

开发基础 Activity

工作流的主要职责之一是编排 Activity 的执行。Activity 是普通函数或方法执行，旨在执行单个明确定义的操作（无论是短时，还是长时运行），比如查询数据库、调用第三方 API 或转码媒体文件。Activity 可以与 Temporal 平台外部进行交互，也可以使用 Temporal Client 与 Temporal 服务进行交互。要让工作流能够执行 Activity，必须定义 Activity 定义。

独立 Activity（Standalone Activity）是指独立运行的 Activity，无需由工作流进行编排。无需在工作流定义中启动 Activity，可以直接从 Temporal 客户端启动独立 Activity。

编写独立 Activity，以及将其注册到 Worker 的方式与工作流 Activity 完全相同。唯一的区别在于，独立 Activity 直接从 Temporal 客户端执行。

可以使用 @activity.defn 装饰器开发 Activity 定义。通过装饰器参数（比如 @activity.defn(name="your_activity")）可以将函数注册为具有自定义名称的 Activity。

说明：
Temporal Python SDK 支持通过多种方式实现 Activity：
使用 asyncio 异步执行
使用 concurrent.futures.ThreadPoolExecutor 同步多线程执行
使用 concurrent.futures.ProcessPoolExecutor 和 multiprocessing.managers.SyncManager 同步多进程执行
在 Python 中，阻塞异步事件循环，会将异步程序转变为串行执行的同步程序，从而完全违背使用 asyncio 的初衷。也可能导致潜在的死锁，以及导致任务无法执行的不可预测行为。调试此类问题可能既困难又耗时，因为定位阻塞调用的来源往往不显而易见。
因此，建议不要在异步 Activity 内部进行阻塞调用，或者使用异步安全的库，执行这些操作。如果必须使用阻塞库，那么考虑改用同步 Activity。

from temporalio import activity
from your_dataobject import YourParams


@activity.defn(name="your_activity")
async def your_activity(input: YourParams) -> str:
    return f"{input.greeting}, {input.name}!"

开发 Activity 参数

Activity 定义支持的参数总数没有明确限制。但最终编码进 gRPC 消息负载（Payload）的数据总大小存在限制。

单个参数的大小上限为 2 MB。gRPC 消息的总大小（包含所有参数）上限为 4 MB。

此外，所有负载数据都将被记录到工作流执行事件历史中，过大的事件历史可能影响 Worker 性能。这是因为整个事件历史可能随着工作流任务被传输到 Worker 进程。

一些 SDK 要求传递上下文对象，而另一些则不需要。对于应用数据（即被序列化，并且编码进负载的数据），建议使用单个对象封装传递给 Activity 的应用数据。这样做的好处是，可以在不破坏函数或方法签名的情况下，更改传递给 Activity 的数据。

Activity 参数是指使用 @activity.defn 装饰的函数的参数。可以是 Temporal 能够转换的任何数据类型，包括带正确类型注解的数据类。从技术上讲，Activity 可以有多个参数，但 Temporal 强烈建议使用包含所有输入字段的单个数据类参数。

from temporalio import activity
from your_dataobjectimport YourParams


@activity.defn(name="your_activity")
asyncdefyour_activity(input: YourParams) ->str:
    return f"{input.greeting}, {input.name}!"

定义 Activity 返回值

Activity 返回的所有数据都必须可序列化。

Activity 的返回值受 Temporal 中负载大小限制的约束。默认的负载大小限制为 2MB，事件历史事务中 gRPC 消息大小的硬性限制为 4MB。注意，所有返回值都被记录到工作流执行事件历史中。

Activity 执行可以返回输入值和其他 Activity 值。

以下示例定义的 Activity 接受对象作为输入，返回字符串。

from temporalio import activity
from your_dataobject import YourParams


@activity.defn(name="your_activity")
asyncdefyour_activity(input: YourParams) ->str:
returnf"{input.greeting},{input.name}!"

自定义 Activity 类型

Activity 具有类型，也被称为 Activity 名称。下面的示例展示如何为 Activity 类型设置自定义名称。

可以在装饰器参数中为 Activity 指定自定义名称。比如 @activity.defn(name="your-activity")。如果未指定 name 参数，则默认使用函数名称。

from temporalio import activity
from your_dataobject import YourParams


@activity.defn(name="your_activity")
async def your_activity(input: YourParams) -> str:
    return f"{input.greeting}, {input.name}!"