Funboost Remote Deploy

Funboost 远程部署（fabric_deploy）

概述

Funboost 提供 函数级 远程部署能力：从本地 Python 代码出发，通过 SSH 自动上传项目代码到 Linux 服务器，并在远程自动导入指定函数、启动消费进程。无需在远程机器安装 git，也无需手动上传代码或登录逐台启动。

核心原则： 部署粒度是 单个 @boost 函数，不是整个 .py 脚本；远程通过 python3 -c "from module import func; func.multi_process_consume(n)" 精确启动指定函数的消费者。

适用场景

• 多台物理机/测试机，不方便用 k8s / CodePipeline 等运维工具
• 需要灵活指定「哪台机器跑哪个函数、开几个进程」
• 本地开发机一键把消费者部署到远程 Linux 并启动
• 与 multi_process_consume 结合，在远程充分利用多核 CPU

不适用场景

• 已有 阿里云 CodePipeline、k8s 一键部署 等成熟运维体系 → 优先用那些，不必用 fabric_deploy
• 远程机器是 Windows → 不支持，仅支持 Linux
• 远程机器 没有 Python 基本环境 → 需先自行安装 Python 及项目依赖

1. 远程部署的核心概念

分布式 + 函数级部署

Funboost 是分布式函数调度框架：消息队列任务在多机间共享，任意机器上的消费者都能消费同一队列。fabric_deploy 在此基础上增加了 代码级自动化：

函数级 vs 脚本级

传统部署是上传脚本后在远程 python worker.py。Funboost 会根据函数所在文件自动推导模块路径，例如：

test_frame/test_fabric_deploy/test_deploy1.py 中的 f2
→ from test_frame.test_fabric_deploy.test_deploy1 import f2
→ f2.multi_process_consume(2)

因此可以 同一文件内不同函数部署到不同机器，或同一机器上只启动部分函数。

导入方式说明

funboost/__init__.py 中 未导出 fabric_deploy（fabric2 尚未适配 Python 3.12+，顶层导入会导致高版本报错）。正确用法：

# ✅ 推荐：@boost 装饰后的函数对象自带 .fabric_deploy 方法
my_task.fabric_deploy(host, port, user, password, process_num=2)

# ✅ 也可直接导入底层函数
from funboost.core.fabric_deploy_helper import fabric_deploy, kill_all_remote_tasks
fabric_deploy(my_task, host, port, user, password, process_num=2)

# ❌ 不要写（高版本 Python 可能报错或不存在）
from funboost import fabric_deploy

2. fabric_deploy 配置方式

基本 SSH 连接参数（必填）

上传控制参数

远程执行参数

invoke_runner_kwargs 常用项

远程目录规则

源码根据 sys.path[1] 解析本地项目根目录名，远程路径为：

• user == 'root' → /codes/{项目目录名}/
• 其他用户 → /home/{user}/codes/{项目目录名}/

3. 函数级粒度部署

每个 @boost 装饰后的函数（Booster 对象）独立调用 .fabric_deploy()，互不影响：

from funboost import boost, BoosterParams, BrokerEnum

@boost(BoosterParams(queue_name="queue_a", broker_kind=BrokerEnum.REDIS_ACK_ABLE))
def task_a(x):
    return x * 2

@boost(BoosterParams(queue_name="queue_b", broker_kind=BrokerEnum.REDIS_ACK_ABLE))
def task_b(x):
    return x + 1

if __name__ == "__main__":
    # 机器 1 只跑 task_a，2 进程
    task_a.fabric_deploy("192.168.1.10", 22, "deploy", "password", process_num=2)

    # 机器 2 只跑 task_b，1 进程
    task_b.fabric_deploy("192.168.1.11", 22, "deploy", "password", process_num=1)

框架会为每个部署生成唯一进程标识：

funboost_fabric_mark__{queue_name}__{func_name}

重新部署时会先 kill 带该 mark 的旧进程，再启动新进程。

4. multi_process_consume 与远程部署结合

fabric_deploy 的 process_num 参数会直接传给远程的 multi_process_consume(process_num)：

• 多进程：每个进程独立消费，充分利用多核 CPU
• 进程内并发：每个进程内部仍按 BoosterParams 的 concurrent_mode / concurrent_num / qps 等配置运行线程/协程

进程数建议： 达到最大 CPU 性能时，进程数 ≈ CPU 核数即可；每个进程内已有并发池，无需盲目开大。

远程实际执行的命令等价于：

export is_funboost_remote_run=1
export PYTHONPATH=/home/user/codes/your_project:$PYTHONPATH
cd /home/user/codes/your_project
python3 -c "from your_module import your_func; your_func.multi_process_consume(2)" \
  -funboostmark funboost_fabric_mark__your_queue__your_func

本地等价写法（不远程时）：

my_task.multi_process_consume(2)   # 或 mp_consume(2)

5. 代码示例

完整示例：发布 + 远程部署消费

参考 test_frame/test_fabric_deploy/test_deploy1.py：

import time
from funboost import boost, BoosterParams, BrokerEnum

@boost(BoosterParams(
    queue_name="queue_test30",
    qps=0.2,
    broker_kind=BrokerEnum.REDIS,
))
def f2(a, b):
    time.sleep(7)
    print(f"{a} + {b} = {a + b}")
    return a + b

@boost(BoosterParams(
    queue_name="queue_test31",
    qps=0.2,
    broker_kind=BrokerEnum.REDIS,
))
def f3(a, b):
    print(f"{a} - {b} = {a - b}")
    return a - b

if __name__ == "__main__":
    # 本地发布消息
    for i in range(20):
        f3.push(i, i * 2)

    # 远程部署 f3：2 进程消费，增量上传最近 1 天修改的文件
    f3.fabric_deploy(
        "106.55.244.xx", 22, "root", "your_password",
        * 24 * 60 * 60,
        file_volume_limit=100 * 1000,
        process_num=2,
    )

生产常驻：本地脚本退出后远程继续运行

my_task.fabric_deploy(
    host="192.168.1.10",
    port=22,
    user="deploy",
    password="secret",
    process_num=4,
    invoke_runner_kwargs={
        "hide": None,
        "pty": False,   # 关键：本地退出后远程进程不退出
        "warn": True,
    },
)

SSH 私钥登录

my_task.fabric_deploy(
    "192.168.1.10", 22, "deploy", password="",  # 密码可留空
    pkey_file_path="/home/me/.ssh/id_rsa",
    process_num=2,
)

部署前设置远程环境变量

my_task.fabric_deploy(
    "192.168.1.10", 22, "deploy", "secret",
    extra_shell_str="export FUNBOOST_CONFIG=/home/deploy/funboost_config.py",
    process_num=2,
)

杀死远程所有 fabric 部署的进程（慎用）

from funboost.core.fabric_deploy_helper import kill_all_remote_tasks

kill_all_remote_tasks("192.168.1.10", 22, "deploy", "secret")

6. 注意事项

环境与依赖

运行 deploy 脚本前

1. 设置 PYTHONPATH 为项目根目录（框架用 sys.path[1] 定位上传目录和模块路径）：

   $env:PYTHONPATH="D:\codes\funboost"
   python deploy_script.py
   

2. 从项目根目录或确保 sys.path[1] 指向项目根 运行部署脚本，否则模块路径推导和上传目录会错误。

3. 远程需存在 funboost_config.py（或框架默认配置），且 broker 连接信息与本地一致。

SSH 与安全

• 密码写在代码中有泄露风险，生产环境优先 pkey_file_path 私钥 或密钥管理
• 确保远程用户有权限写入 /home/{user}/codes/ 或 /codes/
• 防火墙放行 SSH 端口及消息队列端口

进程生命周期

• pty=True（默认）：适合调试，本地 Ctrl+C 或脚本结束会连带终止远程消费
• pty=False：适合生产常驻；部署脚本可立即退出，消费者在远程后台持续运行
• fabric_deploy 在 后台线程 中执行上传与启动，调用后立即返回，不会阻塞主线程

上传优化

• 首次部署用默认 only_upload_within_the_last_modify_time 做全量上传
• 之后可改为 86400（1 天）或更小，只同步近期修改文件，加快部署
• 大文件、日志、.git 等已通过排除规则过滤，勿把二进制大资源放在代码目录

与 Celery 对比

Celery 需要在每台机器手动拉代码并启动 worker；Funboost fabric_deploy 可在 Python 代码层 自动上传 + 自动按函数启动消费，实现函数级精确部署。

速查表

源码位置

相关 Skill

• using-funboost-basics — 基础使用入门
• funboost-broker-selection — Broker 中间件选型