funboost-workflow

# Funboost 工作流编排 ## 概述 Funboost workflow 模块提供类似 Celery 的原语（`chain`、`group`、`chord`），用于编排多步骤任务流水线，但比 Celery 更简单。 **核心原则：** 用 `chain`（串行）、`group`（并行）、`chord`（扇出后聚合）组合任务编排流水线。 ## 适用场景 - 多步骤处理流水线（下载 -> 处理 -> 上传） - 扇出/扇入模式（并行处理，然后汇总） - 任务依赖——步骤 B 需要步骤 A 的结果 - 复杂多步骤工作流 ## 速查表 | 原语 | 模式 | 说明 | |------|------|------| | `chain(a, b, c)` | A -> B -> C | 串行执行 | | `group(a, b, c)` | A // B // C | 并行执行 | | `chord(group(...), callback.s())` | (A // B // C) -> D | 并行后聚合 | | `func.s(*args)` | — | 创建签名（懒任务引用） | | `func.si(*args)` | — | 创建不可变签名（忽略上游结果） | ## 核心代码模式 ```python from funboost import boost, BoosterParams, BrokerEnum from funboost.workflow import chain, group, chord, WorkflowBoosterParams @boost(WorkflowBoosterParams( queue_name="download_queue", broker_kind=BrokerEnum.REDIS_ACK_ABLE, is_using_rpc_mode=True, )) def download(url: str): print(f"下载 {url}") return f"/tmp/{url.split('/')[-1]}" @boost(WorkflowBoosterParams( queue_name="process_queue", broker_kind=BrokerEnum.REDIS_ACK_ABLE, is_using_rpc_mode=True, )) def process(file_path: str, resolution: str = "720p"): print(f"处理 {file_path}，分辨率 {resolution}") return f"{file_path}.{resolution}.mp4" @boost(WorkflowBoosterParams( queue_name="notify_queue", broker_kind=BrokerEnum.REDIS_ACK_ABLE, is_using_rpc_mode=True, )) def notify(results: list, user_id: int): print(f"通知用户 {user_id}：{len(results)} 个文件已就绪") return "done" ``` ## Chain — 串行流水线 ```python download.consume() process.consume() workflow = chain( download.s("https://example.com/video.mp4"), process.s(resolution="1080p"), ) result = workflow.apply() ``` > **注意：** 使用 workflow 前，所有参与的任务函数必须先调用 `consume()` 启动消费。`apply()` 返回最后一步的 `FunctionResultStatus` 对象（非直接值），通过 `.result` 属性取实际返回值。`group.apply()` 返回结果列表。 每个任务的返回值作为下一个任务的第一个参数传入（使用 `.si()` 可忽略上游结果）。 ## Group — 并行执行 ```python process.consume() parallel_tasks = group( process.s("/tmp/video.mp4", resolution="360p"), process.s("/tmp/video.mp4", resolution="720p"), process.s("/tmp/video.mp4", resolution="1080p"), ) result = parallel_tasks.apply() ``` > **注意：** 参与 group 的任务函数需先 `consume()` 启动消费（同一函数只需调用一次）。 所有任务并发运行。 ## Chord — 扇出后聚合 ```python process.consume() notify.consume() workflow = chord( group( process.s("/tmp/v.mp4", resolution=r) for r in ["360p", "720p", "1080p"] ), notify.s(user_id=1001), ) result = workflow.apply() ``` > **注意：** chord 涉及的所有任务函数（group 成员 + 回调）都需先 `consume()` 启动消费。 group 中所有任务并行执行；全部完成后，结果收集为列表传给回调函数。 ## 复杂嵌套流水线 ```python download.consume() process.consume() notify.consume() workflow = chain( download.s("https://example.com/video.mp4"), chord( group( process.s(resolution=r) for r in ["360p", "720p", "1080p"] ), notify.s(user_id=1001), ), ) result = workflow.apply() ``` > **注意：** 嵌套 chain/group/chord 组合时，每个参与的任务函数都需 `consume()` 启动消费。 ## 重要注意事项 1. **`WorkflowBoosterParams`** — 工作流任务推荐用这个（已默认 `is_using_rpc_mode=True` 并注入必要 Mixin） 2. **导入 `funboost.workflow`** — `.s()` / `.si()` 方法仅在导入 workflow 模块后才可用（如 `from funboost.workflow import chain, ...`） 3. **`.s()` 方法** — 创建签名（懒引用），用于工作流组合；`.si()` 为不可变签名，忽略上游结果 4. **统一 broker** — 工作流中的所有任务建议使用相同的 broker 5. **需配置 Redis** — 工作流依赖 RPC（即使用非 Redis broker 也需要 Redis 存储 RPC 结果） 6. **启动消费者** — 每个任务函数都需 `func.consume()` 才能执行 ## 常见错误 | 错误 | 修正 | |------|------| | 工作流任务用 `BoosterParams` | 推荐用 `WorkflowBoosterParams`（已内置 RPC 和 Mixin） | | 忘记 import funboost.workflow | `.s()` 方法需要导入 workflow 模块后才可用 | | 直接调用 `func(args)` 而非 `func.s(args)` | `.s()` 创建签名，不要直接调用函数 | | 工作流中混用不同 broker | 所有工作流任务保持相同 broker | | 没有启动消费者 | 每个任务函数都需要 `func.consume()` 运行 | ## 相关 Skill - `using-funboost-basics` — 基础使用入门 - `funboost-rpc-mode` — 获取任务执行返回值