15. ⚡ Funboost.FaaS：让函数起飞的 Serverless 引擎 🚀

• FaaS = Function as a Service

• 💡 核心定义：funboost.faas 是框架内置的 云原生适配器，将 Python 消费函数 零代码 转化为可被 HTTP 调用的 微服务接口。

• 🚪✨ 它是连接 "后台离线任务" 与 "前台在线业务" 的桥梁。

---

15.1 🌟 架构质变：从 Worker 到 Service

在传统架构中，后台 Worker 往往是孤立的数据黑盒。Funboost.FaaS 让 每一个函数都成为一个服务端点。

对比维度	🐢 传统 Worker	🚀 Funboost FaaS
触发方式	被动等待队列消息	🌐 HTTP 接口主动唤醒
可控性	黑盒运行，外界不知进度	📊 可查询状态、获取结果、暂停/恢复
灵活性	代码耦合，改动需重启 Web	🔥 热插拔，基于元数据驱动，Web 永不重启

---

15.2 💎 核心特性速览

🏷️ 特性	📝 说明
🔌 开箱即用	针对 FastAPI / Flask / Django 提供预置 Router，一行代码 include_router 即可挂载 20+ 管理接口
🔥 无缝热发现	新增/修改消费函数后，Web 网关 无需重启，基于 Redis 元数据自动发现新逻辑
🔄 RPC 同步调用	HTTP 请求 → MQ 消息 → Worker 处理 → 结果返回 HTTP，前端无需轮询
🧬 全自动参数校验	基于 Redis 元数据中的 final_func_input_params_info，发布接口自动校验 JSON Body
⏰ 可编程调度中心	通过 HTTP 接口动态添加 Crontab / 间隔 / 一次性定时任务
📊 全景透视	实时暴露 QPS、积压量、活跃消费者 IP、错误堆栈，轻松对接 Grafana / Prometheus

---

15.3 🌌 应用场景与价值

🎯 维度	💼 应用场景	✨ 价值
🦄 动态架构	热插拔式微服务	Web 服务 永不重启，基于 Redis 元数据自动发现新逻辑
🌉 异构融合	通用语言网关	Java / Go / Node.js 通过 HTTP JSON 直接调用 Python 生态
⚡ 交互模式	HTTP-RPC 桥接	异步变同步，need_result=true 即可等待 Worker 返回结果
🏭 研发效能	零代码后端接口	消灭 Controller 层，写完函数即接口
🕹️ 运维管控	可编程调度中心	API 化运维，HTTP 接口动态增删定时任务
🔭 可观测性	全景数据透视	为 Prometheus / Grafana 提供标准化数据源
🧪 质量保障	生产环境热调试	Postman 直接触发函数，Response 中看到返回值或报错堆栈

---

15.4 🛠️ 核心原理

❌ 传统"硬编码"方式的痛点

每新增一个任务函数都要修改 Web 路由代码并重启服务：

# 🚫 反面教材：硬编码路由 + 每次都要重启
@app.post("/api/do_task")
def do_task(task_type: str, data: dict):
    if task_type == 'sms':
       sms_fun.push(data)
    elif task_type == 'email':
       email_fun.push(data)
    # ... 无限的 elif ...

✅ Funboost 解法：基于元数据的动态代理

funboost.faas 的核心是 SingleQueueConusmerParamsGetter——不依赖本地 import，直接从 Redis 读取队列元数据：

1. 📤 消费者启动时，将函数签名、参数类型、队列配置写入 Redis

2. 📥 Web 端收到请求 /funboost/publish {"queue_name":"sms",...}

3. 🔍 Web 端从 Redis 读取 sms 队列的配置元数据

4. 🏗️ 动态生成 Publisher 对象，完成消息发布

🎯 结果：Web 服务与 Worker 逻辑 彻底解耦，新增函数无需修改 Web 代码。

---

15.5 📚 API 接口清单 (Swagger 📖 开箱即用)

启动服务后，访问 /docs 即可看到全套接口：

15.5.1. 📨 消息发布与结果查询

方法	路径	功能	核心参数
POST	/funboost/publish	🎯 万能发布接口，支持 RPC 模式同步等待结果	queue_name, msg_body, need_result
GET	/funboost/get_result	🔍 根据 task_id 获取异步任务执行结果	task_id, timeout
GET	/funboost/get_msg_count	📊 实时获取队列当前堆积消息数量	queue_name

15.5.2. 🔍 服务发现与元数据 (FaaS 核心 💎)

方法	路径	功能	核心参数
GET	/funboost/get_all_queues	📋 获取项目注册的所有队列名称列表	无
GET	/funboost/get_queues_config	📄 获取所有队列的详细配置（含 final_func_input_params_info）	无
GET	/funboost/get_one_queue_config	🔎 获取指定队列的详细配置信息	queue_name

15.5.3. ⚙️ 队列运维与监控

方法	路径	功能	核心参数
GET	/funboost/get_queue_run_info	📈 单队列透视（消费速率、积压量、活跃消费者 IP）	queue_name
GET	/funboost/get_all_queue_run_info	🌐 全局视角（适合监控大屏）	无
POST	/funboost/pause_consume	⏸️ 暂停某队列消费	queue_name
POST	/funboost/resume_consume	▶️ 恢复某队列消费	queue_name
POST	/funboost/clear_queue	⚠️ 清空队列所有堆积消息（慎用！）	queue_name
DELETE	/funboost/deprecate_queue	🗑️ 从 Redis 移除不再使用的队列名	queue_name

15.5.4. ⏰ 动态定时任务编排

方法	路径	功能	核心参数
POST	/funboost/add_timing_job	➕ 创建定时任务（支持 date/interval/cron）	queue_name, trigger, job_id, kwargs
GET	/funboost/get_timing_jobs	📋 获取定时任务列表	queue_name(可选)
GET	/funboost/get_timing_job	🔎 获取指定任务详细配置	job_id, queue_name
POST	/funboost/pause_timing_job	⏸️ 暂停定时计划	job_id, queue_name
POST	/funboost/resume_timing_job	▶️ 恢复定时计划	job_id, queue_name
DELETE	/funboost/delete_timing_job	🗑️ 永久删除定时任务	job_id, queue_name
DELETE	/funboost/delete_all_timing_jobs	💣 批量清空所有定时任务	queue_name(可选)

15.5.5. 🎛️ Scheduler 调度器控制

方法	路径	功能	核心参数
GET	/funboost/get_scheduler_status	📊 获取调度器状态	queue_name, job_store_kind
POST	/funboost/pause_scheduler	⏸️ 暂停整个调度器	queue_name, job_store_kind
POST	/funboost/resume_scheduler	▶️ 恢复调度器运行	queue_name, job_store_kind

15.5.6. 🏷️ 项目筛选 (care_project_name)

多项目场景下，筛选只显示特定项目的队列信息：

方法	路径	功能	核心参数
GET	/funboost/get_care_project_name	🔍 获取当前筛选设置	无
POST	/funboost/set_care_project_name	🏷️ 设置筛选项目	care_project_name
GET	/funboost/get_all_project_names	📋 获取所有项目名称列表	无

15.5.10. 📄 Swagger 接口文档截图

启动 FastAPI Web 服务后，访问 /docs 即可看到完整接口文档：

---

15.6 💻 快速接入指南

15.6.1 🌟 FastAPI 接入（推荐）

import uvicorn
from fastapi import FastAPI
from funboost.faas import fastapi_router, CareProjectNameEnv

# （可选）设置项目名，只管理本项目相关的队列
CareProjectNameEnv.set('my_awesome_project')

app = FastAPI()

# ✨ 核心一行代码：注册路由
app.include_router(fastapi_router)

if __name__ == '__main__':
    # 🌐 启动后访问 http://127.0.0.1:8000/docs 即可看到 Funboost 管理接口
    uvicorn.run(app, host="0.0.0.0", port=8000)

15.6.2 🧪 Flask 接入

from flask import Flask
from funboost.faas import flask_blueprint

app = Flask(__name__)

# ✨ 核心一行代码：注册蓝图
app.register_blueprint(flask_blueprint)

if __name__ == '__main__':
    app.run(port=5000)

15.6.3 🎯 Django 接入 (Django-Ninja)

from ninja import NinjaAPI
from funboost.faas import django_router

api = NinjaAPI()

# ✨ 核心一行代码：添加路由
api.add_router("/funboost", django_router)

# urls.py
urlpatterns = [
    path("api/", api.urls),
]

15.6.4 🔒 权限控制 (Security)

像控制普通接口一样控制 Funboost 接口：

# FastAPI 示例：添加 Token 验证
from fastapi import Depends, HTTPException, Header

async def verify_token(token: str = Header(...)):
    if token != "secret-password":
        raise HTTPException(status_code=400, detail="Invalid Token")

# 🔐 给所有 Funboost 接口加上权限验证
app.include_router(fastapi_router, dependencies=[Depends(verify_token)])

---

15.7 🎯 调用示例：通过 HTTP 发布 RPC 任务

假设有消费函数 @boost(BoosterParams(queue_name='test_funboost_faas_queue', is_send_consumer_heartbeat_to_redis=True)) def add(x, y): return x + y：

📤 请求：

curl -X POST "http://127.0.0.1:8000/funboost/publish" \
-H "Content-Type: application/json" \
-d '{
  "queue_name": "test_funboost_faas_queue",
  "msg_body": {"x": 33, "y": 44},
  "need_result": true, 
  "timeout": 10
}'

📥 响应（Funboost 自动计算并返回）：

{
  "succ": true,
  "msg": "获取成功",
  "data": {
    "task_id": "test_funboost_faas_queue_result:3aaf05e6-...",
    "status_and_result": {
      "queue_name": "test_funboost_faas_queue",
      "function": "add",
      "params": {"x": 33, "y": 44},
      "result": 77,
      "success": true,
      "time_cost": 1.006,
      "run_times": 1,
      "host_name": "LAPTOP-7V78BBO2",
      "script_name": "start_consume.py"
    }
  },
  "code": 200
}

---

⚠️💡 15.8 注意事项

1. ⚠️ 必须设置 is_send_consumer_heartbeat_to_redis=True
   使用 FaaS 功能时，@boost 一定需要设置此参数，否则无法从 Redis 获取 queue_name 的配置元数据和运行信息。

2. 🔄 别忘了启动消费
   funboost.faas 只是 Web 网关，仍需启动 funboost 消费者来实际执行任务。消费可以和 Web 一起启动，也可以单独部署。（因为 funboost.faas 基于 Redis 注册的元数据驱动，不需要 import 具体的消费函数。）

---

🧪 15.9 funboost.faas 演示例子

🌐 example_faas 完整示例

# 1️⃣ 启动 funboost faas web
python example_fastapi_faas.py

# 2️⃣ 启动消费
python start_consume.py

# 3️⃣ 访问接口文档
http://127.0.0.1:8000/docs

# 4️⃣ 测试请求 faas 接口
python example_req_fastapi.py

15.9.2 🧪 如何测试函数自动发现？

1. 📝 新写一个 @boost 函数并部署启动消费

2. 🌐 Web 无需重启，直接通过 /funboost/publish 即可调用新函数

3. 🛡️ 乱造请求（不存在的 queue_name 或错误的 msg_body），接口会迅速拒绝并反馈错误原因

4. 📦 Web 服务和消费函数可分开写在不同的 Git 项目中

---

⏰ 15.10 funboost.faas 定时任务管理 —— 架构级解耦的调度中心 🏗️

💡 核心思想：凡是 Web 管理端需要 import 业务函数的调度系统，都不可能成为真正的通用调度平台。

Funboost.FaaS 内置 动态定时任务管理中心，一个 Web 界面即可管理任意多个 funboost 项目的定时任务增删改查。Web 端完全不需要 import 任何业务代码。

15.10.1 🥊 核心差异：为什么优于传统 APScheduler 管理？

传统 Flask-APScheduler / Django-APScheduler 存在 强耦合 问题：

😫 传统痛点	📝 说明
🔗 无法物理分离	Web 管理后台必须和业务代码在同一个项目中，或安装所有业务依赖
💥 牵一发动全身	修改消费函数代码，Web 管理端可能也需重部署
🚫 不支持跨项目	无法用一个统一管理台管理多个 Git 仓库的定时任务

📌 根本原因：传统 APScheduler 底层依赖 pickle 序列化函数对象，或需通过字符串路径在本地寻找函数。

---

15.10.2 🚀 Funboost 的解决方案：彻底解耦

哲学：基于 "队列名 + JSON 数据" 的调度，而非基于"函数代码"的调度。

🏷️ 角色	📝 职责
🌐 Web 端（管理面）	只往 Redis JobStore 写入调度记录。完全不需要 import 业务代码。甚至可以是 Go / Java 前端
⚙️ Worker 端（执行面）	扫描 Redis，触发时间到了就推消息到队列，消费者执行真正的函数

🏆 结果：Web 管理界面和定时任务执行函数，可以处于 完全隔离的 Git 项目，部署在不同服务器，互不依赖。

---

15.10.3 🛠️ 接口能力概览

所有接口支持 Redis 持久化存储。即使 Web 挂了或消费者重启，定时任务依然存在于 Redis 中。

🎯 功能	方法	路径	📝 描述
➕ 添加任务	POST	/add_timing_job	基于队列名添加，支持 Interval/Cron/Date 触发器
📋 查询任务	GET	/get_timing_jobs	查看所有任务列表
🔎 任务详情	GET	/get_timing_job	获取单个任务详细参数
⏸️ 暂停任务	POST	/pause_timing_job	暂停调度（不发消息）
▶️ 恢复任务	POST	/resume_timing_job	恢复调度
🗑️ 删除任务	DELETE	/delete_timing_job	永久删除
⏸️ 调度器控制	POST	/pause_scheduler	暂停整个调度器

15.10.4 💻 调用示例：完全黑盒化的管理

项目 A（Git Repo A）运行消费函数：

from funboost import boost, BoosterParams, BrokerEnum, ApsJobAdder

@boost(BoosterParams(queue_name='send_email_queue', broker_kind=BrokerEnum.REDIS,
                     is_send_consumer_heartbeat_to_redis=True))
def send_email(email_address, content):
    print(f"Sending email to {email_address}: {content}")

if __name__ == '__main__':
    ApsJobAdder(send_email, job_store_kind='redis', is_auto_start=True)
    send_email.consume()

项目 B（Git Repo B）或 Postman 中，完全不知道 consumer.py 代码，远程管理定时任务：

1️⃣ 动态添加任务（无需代码引用）

只需要知道队列名 send_email_queue，即可添加定时计划：

import requests

url = "http://faas-admin-server:8000/funboost/add_timing_job"

payload = {
    "queue_name": "send_email_queue",
    "job_id": "daily_report_job",
    "trigger": "cron",
    "hour": "9",
    "minute": "0",
    "kwargs": {
        "email_address": "boss@example.com", 
        "content": "Daily Report"
    },
    "job_store_kind": "redis",
    "replace_existing": True
}

resp = requests.post(url, json=payload)
print("✅ 任务添加成功，Project A 的 Worker 将自动感知并执行")

2️⃣ 管理任务

# ⏸️ 暂停任务
requests.post("http://faas-admin-server:8000/funboost/pause_timing_job", 
              params={"job_id": "daily_report_job", "queue_name": "send_email_queue"})

# 🗑️ 删除任务
requests.delete("http://faas-admin-server:8000/funboost/delete_timing_job", 
                params={"job_id": "daily_report_job", "queue_name": "send_email_queue"})

15.10.5 ⚙️ 技术原理

🏷️ 要点	📝 说明
🔗 中间件化	利用 Redis 作为中间媒介（JobStore），Web 端只写 JSON 配置数据，不涉及 Pickle 序列化
🔄 执行逻辑转换	传统：Trigger → 执行函数 Funboost：Trigger → Push msg → Consumer 执行函数
🔒 分布式锁	Worker 取任务时加分布式锁，即使部署 10 节点也只触发一次推送

15.10.6 📊 总结对比

🏷️ 对比	📝 传统 APScheduler	✨ Funboost FaaS
耦合度	🔗 强耦合（必须在一个项目中）	✂️ 零耦合（完全独立部署）
扩展性	😫 难以管理大型微服务架构	🏢 一个管理中心管理全公司所有项目的定时任务

---

15.11 🆚 funboost.faas vs 传统 Web 框架写接口

💡 本质区别：Django/Flask 以 "请求-响应" 为中心，funboost 以 "函数" 为中心。函数天然可复用，所以不需要适配层！

15.11.1 🐢 传统 Django/Flask 的痛点

Django 的视图函数一般不直接写复杂逻辑，因为视图函数不能作为普通函数被复用。开发者被迫多写一层"搬运工"：

# 🐢 视图函数 - 只是个"搬运工"，不能直接复用
@api_view(['POST'])
def calculate_score_view(request):
    user_id = request.data['user_id']
    weights = request.data['weights']
    result = calculate_score(user_id, weights)  # 被迫多一层调用
    return Response({'result': result})

# 真正的业务逻辑 - 另外封装
def calculate_score(user_id, weights):
    # 复杂逻辑...
    return score

😫 问题：每个功能要写两遍（业务函数 + 视图适配器） + 配路由 + 写序列化器 + 写参数校验

---

15.11.2 🚀 funboost.faas 的设计哲学：函数即接口

# ✨ 这就是业务函数，同时也是 HTTP 接口，也能被其他代码直接调用
@boost(BoosterParams(queue_name="calculate_score", is_send_consumer_heartbeat_to_redis=True))
def calculate_score(user_id: int, weights: dict):
    # 复杂逻辑...
    return score

# 🔹 直接当普通函数调用
result = calculate_score(123, {"a": 0.5})

# 🔹 通过队列异步调用
calculate_score.push(123, {"a": 0.5})

# 🔹 通过 HTTP 接口调用
# POST /funboost/publish {"queue_name": "calculate_score", "msg_body": {...}}

🎯 一个函数，三种调用方式，没有中间适配层！

---

15.11.3 📊 代码量对比

功能点	🐢 Django 需要写	🚀 funboost 需要写
业务函数	✅ 1份	✅ 1份
视图/路由	❌ 额外1份	✨ 0（自动）
序列化器	❌ 额外1份	✨ 0（自动）
参数校验	❌ 额外写	✨ 0（根据函数签名自动）
接口文档	❌ 额外写	✨ 0（自动生成）

---

15.11.4 🔄 上新功能流程对比

对比维度	🐢 传统 Django/Flask	🚀 funboost.faas
上新流程	写视图 → 配路由 → 写序列化 → 参数校验 → 重启服务	写 @boost 函数 → 部署 → 自动可调用 ✨
接口文档	需手写或用 Swagger 注解	📄 自动从函数签名生成
参数校验	手动写校验逻辑	🧬 自动根据 final_func_input_params_info 校验
Web 服务重启	❌ 每次都要重启	✅ 永不重启（热发现）
跨项目复用	需打包成库或微服务	🌐 共享 Redis 即可跨项目调用

---

15.11.5 🎉 最爽的几个点

15.11.5.1 ✨ 真正的"写完即上线"

# 只写这个函数，部署上线后，HTTP 接口马上就能调用
@boost(BoosterParams(
    queue_name="new_feature",
    is_send_consumer_heartbeat_to_redis=True  # ⚠️ faas 功能必须设置此参数
))
def calculate_score(user_id: int, weights: dict):
    return score

15.11.5.2 🌐 Web 网关 = 万能入口

一个 app.include_router(fastapi_router) 搞定所有接口，不用再纠结：

• 🤔 这个接口用 GET 还是 POST？

• 🤔 URL 路径怎么设计？

• 🤔 参数放 query 还是 body？

15.11.5.3 🔄 天然支持异步和 RPC

传统视图函数要实现"提交任务 → 轮询结果"需要额外设计，funboost 直接内置：

# ✨ need_result=True 一行搞定 RPC
{"queue_name": "xxx", "msg_body": {...}, "need_result": true}

15.11.5.4 🤝 跨团队协作超方便

其他团队只需知道 queue_name（入参格式能通过 FaaS 接口传递 queue_name 获取），就能直接调用你的功能：

• 🌍 不用关心你用什么语言实现

• 🖥️ 不用关心你的服务部署在哪里

• 🛡️ 不用担心你的服务有没有挂（消息队列会等你恢复）

---

15.11.6 ⚖️ 什么场景传统方式更合适？

🏷️ 场景	推荐方式	📝 理由
精细控制 HTTP 状态码/headers	🐢 传统视图函数	框架原生能力，控制力更强
实时流式响应（SSE/WebSocket）	🐢 传统视图函数	需要长连接或特定协议
复杂中间件链条	🐢 传统视图函数	依赖特定 Web 框架生态
CPU 密集型异步任务	🚀 funboost.faas ✅	不阻塞 Web 主线程
IO 密集型异步任务	🚀 funboost.faas ✅	原生支持异步高并发
跨服务编排调用	🚀 funboost.faas ✅	队列解耦，天然分布式
快速迭代上新功能	🚀 funboost.faas ✅	写完即发，无需重启

---

15.11.7 💡 本质总结

🎯 Django/Flask 以"请求-响应"为中心，funboost 以"函数"为中心。

函数天然可复用，所以不需要适配层！这就是 "函数即服务" (FaaS) 的魅力 ——
✨ 专注业务逻辑本身，基础设施全自动化。