Funboost Faas Deploy

Funboost FaaS 微服务部署

概述

Funboost 提供内置的 HTTP Router（FastAPI、Flask、Django），一行代码即可为所有 boost 装饰的函数暴露发布和结果查询接口。

核心原则： 不要手动编写发布/结果 API — 直接使用内置 router。

适用场景

• 将任务函数暴露为 HTTP 接口
• 构建基于分布式队列的微服务 API
• 将 funboost 集成到现有 Web 应用
• 为前端/其他服务创建任务提交 API

速查表

FastAPI 集成

前置条件： FaaS 接口依赖 Redis 发现已注册队列的元数据。需确保消费者已启动且通过心跳注册到 Redis，并在 funboost_config.py 中配置 Redis 连接。

from fastapi import FastAPI
from funboost.faas import fastapi_router
from funboost import boost, BoosterParams, BrokerEnum

app = FastAPI()
app.include_router(fastapi_router)

@boost(BoosterParams(
    queue_name="email_queue",
    broker_kind=BrokerEnum.REDIS_ACK_ABLE,
    is_using_rpc_mode=True,
))
def send_email(to: str, subject: str, body: str):
    print(f"发送邮件到 {to}")
    return {"status": "sent", "to": to}

# 消费者在单独的进程/脚本中启动：
# send_email.consume()

自动提供以下接口：

• POST /funboost/publish — 提交任务
• GET /funboost/get_result?task_id=xxx&timeout=5 — 查询任务结果

请求体格式（POST /funboost/publish）：

{
    "queue_name": "email_queue",
    "msg_body": {"to": "user@example.com", "subject": "Hello", "body": "Hi"},
    "need_result": true,
    "timeout": 60
}

Flask 集成

from flask import Flask
from funboost.faas import flask_blueprint

app = Flask(__name__)
app.register_blueprint(flask_blueprint)

# 正常定义 @boost 任务...

发布端与消费端分离

发布端（Web 服务器）：

# web_app.py
from fastapi import FastAPI
from funboost.faas import fastapi_router

app = FastAPI()
app.include_router(fastapi_router)

# 注意：FaaS router 默认从 Redis 元数据动态发现队列
# 仅当设置环境变量 os.environ['funboost.faas.is_use_local_booster'] = 'true' 时才需要本地导入任务模块

消费端（Worker 进程）：

# worker.py
from tasks import send_email, process_order

send_email.consume()
process_order.consume()

自定义 API 接口（手动方式）

如果内置 router 不满足需求：

from fastapi import FastAPI
from funboost import boost, BoosterParams, BrokerEnum

app = FastAPI()

@boost(BoosterParams(
    queue_name="custom_task",
    broker_kind=BrokerEnum.REDIS_ACK_ABLE,
    is_using_rpc_mode=True,
))
def heavy_computation(data: list):
    return sum(data)

@app.post("/compute")
async def submit_computation(data: list[int]):
    result = await heavy_computation.aio_push(data)
    return {"task_id": result.task_id}

@app.get("/compute/{task_id}")
async def get_result(task_id: str):
    from funboost.core.msg_result_getter import AioAsyncResult
    aio_result = AioAsyncResult(task_id)
    status_dict = await aio_result.status_and_result
    return {"success": status_dict.get("success"), "result": status_dict.get("result")}

常见错误

相关 Skill

• funboost-rpc-mode — 获取任务执行返回值
• funboost-funweb-ops — Web 管理界面运维