华为云函数工作流入门实操：用 FunctionGraph 搭一个定时清理与通知小流程

很多刚接触 Serverless 的同学，第一反应是“函数就是一段代码”，其实函数工作流的价值在于把多个函数、判断和等待串成一条可观测、可重试的流水线。本文以一个最常见的运维小场景为例：每天定时清理过期临时文件，清理完成后把结果通知到运维群组。整个流程不需要常驻服务器，也不需要自己搭调度器。

一、先把流程画清楚，再动手

在动手之前，先把流程拆成三步：

1. 定时触发：每天固定时间启动一次；
2. 清理函数：扫描指定目录或对象列表，删除超过保留天数的文件，并输出删除数量；
3. 通知函数：把清理结果汇总成一条消息，发送到通知渠道。

这样拆的好处是每一步都可以单独测试、单独改动。清理逻辑变了，不用动定时和通知；通知渠道变了，也不用动清理逻辑。

二、准备一个清理函数

清理函数的输入建议只包含三个字段：目标路径、保留天数和是否真正删除。保留“是否真正删除”这个开关非常重要，第一次上线先用演练模式跑，只输出“将要删除哪些文件”，确认无误后再打开真正删除。

一个稳妥的清理函数通常要做到四点：

• 只删明确匹配规则的文件，不用模糊通配；
• 删除前记录文件路径、大小和最后修改时间；
• 删除失败不中断整批任务，而是把失败项单独汇总；
• 返回结构化的结果，例如成功数、失败数和失败清单。

def clean_expired_files(path, keep_days, dry_run=True):
    # 伪代码：按最后修改时间判断是否过期
    expired = [f for f in list_files(path) if is_expired(f, keep_days)]
    if dry_run:
        return {"to_delete": len(expired), "files": expired}
    deleted, failed = delete_files(expired)
    return {"deleted": deleted, "failed": failed}

三、用定时触发器启动流程

函数工作流的定时触发一般有两种写法：按 Cron 表达式定时触发，或者由业务事件触发。对清理任务来说，Cron 是最直观的选择，例如每天凌晨业务低峰期执行一次。

配置时注意三点：第一，时区要和业务时间一致，避免凌晨任务跑到白天；第二，第一次上线把执行时间错开业务高峰；第三，开启执行记录，方便事后核对“昨天到底删了多少”。

四、把结果接进通知函数

清理完成后，工作流把上一步的返回值传给通知函数。通知内容不需要很长，但要包含四个要素：执行时间、目标路径、成功数量和失败清单摘要。如果有失败项，通知里要明确写出“需要人工复核”，而不是只报一句成功。

通知函数本身也要做好失败隔离：通知发送失败不应该反过来影响清理结果的记录。正确的做法是清理结果先落盘或写入日志，通知只是对结果的再投递。

五、失败了怎么办：重试与兜底

函数工作流比“写一个定时脚本”多出来的价值，主要在失败处理上。建议按下面三层来设计：

1. 单步重试：网络抖动、临时限流这类问题，重试一两次往往就能恢复；
2. 分支兜底：清理函数多次失败后，走人工复核分支，而不是悄悄跳过；
3. 全局可观测：每一次执行都要留下执行 ID、开始时间、结束时间和每一步的输出，出了问题先查执行记录，再改代码。

六、一个最小可用的检查清单

上线前，对照下面这份清单自查一遍：

• [ ] 演练模式已跑过，删除清单已人工确认；
• [ ] 保留天数、目标路径都写在配置里，而不是写死在代码里；
• [ ] 定时触发的时区和执行时间已核对；
• [ ] 成功和失败都有通知，失败项可追溯；
• [ ] 执行记录可查询，单步失败可单独重试。

七、总结

函数工作流入门的关键不是一次写出很复杂的流程，而是先把“定时触发、清理、通知”这条最小链路跑通，再逐步加上重试、分支和告警。流程越小，越容易验证；验证通过后，再把更多运维任务按同样的模式接进来，收益会越滚越大。

下一步可以试着把这个小流程扩展一下：清理前先做一次容量统计，清理后对比前后容量变化，把“省了多少空间”也写进通知里，运维价值会更直观。