6. 常见问题回答

6.0 "funboost 值得学习使用吗？"

绝对值得。 核心优势一览：

零侵入：函数上加 @boost 即可，不改项目结构，去中心化（无 Celery 那样的 app 实例）
近 50 种消息队列 + 5 种并发模式 + 30+ 种任务控制能力
性能：是 Celery 的 22~46 倍（控制变量法实测）
三位一体：异步任务队列 + RPC + FaaS 微服务
生产级治理：ACK、重试、死信、熔断、OpenTelemetry 追踪、funweb 管理后台
极简 API：核心就是 @boost + BoosterParams，IDE 全补全，无需命令行

详见 6.0b（做什么）、6.0c（难吗）、6.0d（场景）、6.0e（替代）、6.0f（超级装饰器）。

6.0b funboost 是做什么的？

一句话：Funboost 是一个自带 FaaS 能力的通用分布式函数调度框架。加一行 @boost，函数立刻拥有分布式、高并发、高可靠能力；通过 funboost.faas，瞬间变成 HTTP 微服务接口。

你的痛点	funboost 怎么解决
需要给 Java/Go 调用 Python 函数	`funboost.faas` 自动变成 HTTP 接口
任务处理太慢/程序卡死	`@boost` 自动多线程/协程/多进程/多机器并发
担心崩溃/断电丢任务	ACK + 自动重试 + 死信队列
精确控制执行频率	`qps` 参数，从 0.001 到 50000

适用场景：FaaS 微服务、AI 推理、ETL、分布式爬虫、消息通知、CDC 实时流、物联网。

设计哲学：原子化（函数为核心） + 服务化（funboost.faas） + 赋能而非奴役（零侵入）。

6.0c funboost 学习难吗？

不难。 教程长是因为功能全面、文档完备，而非框架复杂。核心只需掌握三件事：

@boost 装饰器：给函数加上它
BoosterParams 类：配置任务参数
.push() 和 .consume()：发布和消费

from funboost import boost, BrokerEnum, BoosterParams

@boost(BoosterParams(queue_name="task_queue_name1", qps=5, broker_kind=BrokerEnum.SQLITE_QUEUE))
def task_fun(x, y):
    print(f'{x} + {y} = {x + y}')

if __name__ == "__main__":
    for i in range(100):
        task_fun.push(i, y=i * 2)
    task_fun.consume()

所有功能（切换消息队列、设置并发模式、启用 30+ 种控制）都只是修改 BoosterParams 里的参数。

学习路径：

跑通 1.3.1 的最简例子（5 分钟）
修改 broker_kind 和 qps 感受效果
按需查阅对应章节

6.0d 可以用于哪些场景？（7 大核心领域）

领域	场景举例	核心优势
异步后台任务	发邮件、生成报表、视频转码	配置简单，支持 Windows，精准控频
FaaS 微服务	跨语言调用、前后端交互、热插拔服务	写完函数即服务，无需 Web Controller
分布式爬虫	全网抓取、竞对监控、应对反爬	ACK 不丢数据，多机多进程叠加并发
ETL 与 CDC	MySQL Binlog 监听、Kafka 日志清洗	一行代码实现 Binlog 监听
定时任务	日结报表、订单超时取消	分布式锁防重复，支持 Web 管理
物联网	MQTT 传感器消息入库、设备控制	直接对接 IoT 协议
运维自动化	一键部署到 100 台机器	fabric_deploy 集成

6.0e 能替代哪些框架？

替代对象	funboost 优势
Celery / RQ / Huey / Dramatiq	性能 46 倍、配置简单、无 Broker 限制、支持 Windows
Scrapy / Feapder / PySpider	编程自由、天然分布式、ACK 断点续传
Nameko / gRPC（部分场景）	funboost.faas 写完函数即服务，无需 .proto 文件
APScheduler / Crontab / Celery Beat	分布式锁防重复、Web 管理界面
Canal / Logstash（轻量场景）	Python 技术栈一个脚本搞定 MySQL CDC

6.0f 为什么可以将 funboost 看作超级装饰器？

即使不用分布式和消息队列，使用 MEMORY_QUEUE 模式，一个 @boost 就能抵常规 10 个装饰器叠加：

功能	关键参数
并发控制	`concurrent_num=10`
QPS 控频	`qps=5`（支持小数）
自动重试	`max_retry_times=3, is_using_advanced_retry=True`
任务去重	`do_task_filtering=True, task_filtering_expire_seconds=3600`
函数超时	`function_timeout=10`
消息过期	`msg_expire_seconds=3600`
cron 时间限制	`allow_run_time_cron='* 9-17 * * 1-5'`
asyncio 并发	`concurrent_mode=ConcurrentModeEnum.ASYNC`
自动启动消费	`is_auto_start_consuming_message=True`

综合示例——一个装饰器同时具备并发+控频+重试+去重+超时：

@boost(BoosterParams(
    queue_name="super_task",
    broker_kind=BrokerEnum.MEMORY_QUEUE,
    concurrent_num=10,
    qps=5,
    max_retry_times=3,
    do_task_filtering=True,
    function_timeout=30,
))
def process_data(data: str):
    pass

总结：使用 MEMORY_QUEUE 无需安装 Redis/RabbitMQ，一个装饰器搞定 10+ 种控制，解决多装饰器叠加顺序难题。详细代码示例见第 4 章。

6.1 和 celery 有什么区别？

详见第 2 章的全面对比。简要概括：funboost 在性能（46 倍）、用户编码量、使用难度、IDE 补全、Windows 支持、中间件数量、并发模式、QPS 控频精度等近 40 个维度上全面优于 Celery。

6.2 为什么包名这么长？

为了直接表达框架含义。现在 IDE 都能自动补全，名字长没关系。生产消费模式不是 celery 专利，不必用水果取名。

6.4 框架如何实现定时？

使用定时发布任务实现定时消费。框架的 ApsJobAdder 轻度封装了 apscheduler 包，用户主要学习 apscheduler 的用法即可。

6.5 为什么是函数调度而非类调度？

说明：2024.06 月以后已新增支持实例方法和类方法作为消费函数（见 4.32 章节）。

技术原因：

参数歧义：类实例化需传参、方法也要传参，框架无法确定中间件里的参数哪些传给构造方法、哪些传给普通方法。
幂等性要求：分布式任务要求无状态和幂等，函数天然满足，而类方法依赖对象状态。
序列化代价：如果要发布 self 到消息队列，需 pickle 序列化整个对象。pickle 消耗 CPU 大、消息体积大，且大量对象（含 lock、redis 连接等属性）根本无法 pickle 序列化。

解决方式：

# 如需调度类的方法，写一个包装函数即可
def your_task(x, y):
    return A(x).add(y)

Celery 作者对此问题的回答完全一致（详见 StackOverflow）。

6.6 怎么调度一个函数的？

基本原理：从消息中间件取出参数 {"a":1,"b":2}，然后执行 add(**{"a":1,"b":2})。

6.7 框架适用哪些场景？

见 6.0d 章节的 7 大核心领域详细说明。简要概括：分布式、并发、控频、断点接续、定时、消费确认、重试、超时杀死、任务过滤、任务过期等数十种功能，只需其中某一种就可以使用。适合一切耗时的函数，不管是 CPU 密集型还是 IO 密集型。

6.8 怎么引入框架？门槛高不高？

先写好你自己的业务函数
在函数上加 @boost 装饰器
执行 your_function.consume() 启动消费

即使不再使用框架，你写的业务函数代码依然完整可用，不需要去掉 @boost 装饰器，函数也能正常直接调用。

6.9 怎么写框架？

需要学习真 OOP 和 36 种设计模式。设计模式的核心价值在于：避免为每种中间件和并发模式做全盘复制粘贴。funboost 通过模板模式 + 工厂模式，使新增中间件类型只需几十行代码。

6.10 框架能做什么？

你在函数里面写什么，框架就是自动高可靠分布式并发做什么。框架在你的函数上加了自动使用消息队列、分布式、多进程+多线程(协程)超高并发、QPS 控频、自动重试等能力。做什么任务是你的函数内部代码决定的。

只要你的代码涉及使用并发（线程/线程池/asyncio），就可以使用此框架。不适用的场景：函数 1 微秒就能完成（如 print("hello")），无需分布式调度。

6.11 日志颜色/详细度调整

6.11.1 控制发布者消费者日志级别

@boost 设置 log_level=20（即 logging.INFO），框架就不再记录每次运行的函数入参和结果。

6.11.2 屏蔽启动时的提示信息

修改 funboost_config.py：

class FunboostCommonConfig(DataClassBase):
    SHOW_HOW_FUNBOOST_CONFIG_SETTINGS = False
    FUNBOOST_PROMPT_LOG_LEVEL = logging.INFO
    KEEPALIVETIMETHREAD_LOG_LEVEL = logging.INFO

6.11.3 控制台颜色调整

使用 pycharm 时建议：file → settings → Editor → Color Scheme → Console Colors 选择 monokai，并自定义 6 个颜色（Blue=1585FF，Cyan=06B8B8，Green=05A53F，Magenta=ff1cd5，Red=FF0207，Yellow=FFB009）。

在项目根目录 nb_log_config.py 设置 DISPLAY_BACKGROUD_COLOR_IN_CONSOLE = False 可去掉背景色块。

6.12 funboost 和 celery 的关系

6.12.2 生产者-broker-消费者不是 celery 的专利

funboost 从 while 1: redis.blpop() 这个第一性原理模型开始迭代，与 celery 没有代码级关系。
两者唯一的共同点是"生产者-消费者"编程模式——这是计算机科学最基础的设计模式，远早于 celery。
任何线程池都是"生产者-broker-消费者"模式（submit → work_queue → worker 线程），这不是 celery 专利。

6.12.3 Gemini-2.5-Pro 大模型分析结论

将 funboost 全部源码上传给 Gemini-2.5-Pro 分析后的结论：

funboost 并非抄袭 Celery 的产物，其核心架构、设计哲学和具体实现与 Celery 存在根本性差异。

关键差异：

维度	funboost	Celery
架构	去中心化，每个函数独立 Booster	中心化 `app` 实例
Broker 操作	直接使用各 MQ 客户端库	依赖 Kombu 重度抽象层
并发池	自研 FlexibleThreadPool（智能伸缩）	依赖 billiard/gevent
Redis ACK	消费者心跳 + 主动健康检查	BRPOPLPUSH + 可见性超时
证据	能将 Celery 整体作为自己的一个 Broker	—

6.12.4 仅用几个 celery 文件证明抄袭 celery 有多困难

celery 源码（含 kombu/billiard）极其复杂。仅 kombu/transport/redis.py 一个文件就 1400+ 行，代码高度抽象难以理解。花几天分析这些源码的时间，从零用 pyredis 自己写更快。celery/worker/consumer/consumer.py 需来回跳转十几个文件和类才能分析，且无类型注释、IDE 无法跳转。

6.13 一个项目中如何连接多个同种类中间件的不同 IP？

方式一：在脚本中调用 patch_frame_config() 设置连接信息。

方式二：将 funboost_config.py 分别复制到不同文件夹，利用 Python 的 import 优先级机制，脚本会自动优先使用同文件夹层级的配置。

方式三：在`funboost_config.py` 中中间件配置使用os.getenv获取环境变量，你运行代码之前设置环境变量。

6.14 什么是确认消费？为什么重要？

启动消费后，任意时刻强制关闭重启脚本，只要函数没有完整执行完成，消息就不会丢失。支持 ACK 的 broker_kind 包括：REDIS_ACK_ABLE、REIDS_ACK_USING_TIMEOUT、REDIS_PRIORITY、REDIS_STREAM、RedisBrpopLpush、RABBITMQ 系列、NATS_JETSTREAM 等。

如果不需要确认消费，使用 BrokerEnum.REDIS（no ack 模式）。

6.15 如何等待队列消息全部消费完成？

启动消费后，调用 wait_for_possible_has_finish_all_tasks 阻塞等待队列消息全部处理完毕：

f.consume()
f.wait_for_possible_has_finish_all_tasks(minutes=3)
print("over")  # 队列全部消费完才会打印

6.16 函数上加多个装饰器

6.16.1 方式一：通过 consuming_function_decorator 传参

推荐方式，可正常使用 push 发布：

from funboost import boost, BoosterParams
from functools import wraps

def incr_deco(redis_key):
    def _inner(f):
        @wraps(f)
        def __inner(*args, **kwargs):
            result = f(*args, **kwargs)
            # 统计逻辑
            return result
        return __inner
    return _inner

@boost(BoosterParams(queue_name='test_queue_235', consuming_function_decorator=incr_deco('my_key')))
def fun(xxx, yyy):
    print(xxx + yyy)
    return xxx + yyy

6.16.2 方式二：直接叠加装饰器 + should_check_publish_func_params=False

需使用 publish 发布（非 push）：

@BoosterParams(queue_name='test_queue_23b', should_check_publish_func_params=False)
@incr_deco('test_run_count')
def fun(xxx, yyy):
    print(xxx + yyy)
    return xxx + yyy

# 发布时必须用 publish
fun.publish({'xxx': 1, 'yyy': 2})

6.17 日志记录太详细？

6.17.a 设置发布者消费者的日志级别

@boost 设置 log_level=20（即 logging.INFO），框架就不再记录每次运行的函数入参和结果。注意 log_level 只控制消费者和发布者命名空间自身的日志，不影响其他命名空间。

6.17b 嫌 funboost 启动时候打印太多提示用户的消息？

主要是提示配置文件路径和内容。老手可通过配置屏蔽：

class FunboostCommonConfig(DataClassBase):
    SHOW_HOW_FUNBOOST_CONFIG_SETTINGS = False
    FUNBOOST_PROMPT_LOG_LEVEL = logging.INFO
    KEEPALIVETIMETHREAD_LOG_LEVEL = logging.INFO

6.18 为什么要设置 PYTHONPATH？

funboost 通过 import funboost_config 读取配置。为了让 Python 在任何工作目录下都能正确导入项目根目录的配置文件，需要设置 PYTHONPATH。

一句话解决：

# Linux
export PYTHONPATH=/home/xiaomin/myproj/; python3 dir2/dir3/run.py

# Windows CMD
set PYTHONPATH=D:\codes\myproj\ & python dir2\dir3\run.py

# Windows PowerShell
$env:PYTHONPATH="D:\codes\myproj\"; python dir2\dir3\run.py

详细说明：https://github.com/ydf0509/pythonpathdemo

6.18.2 为什么 celery scrapy django 不需要用户设置 pythonpath？

因为这些框架固定死了项目目录结构，且运行起点在项目根目录第一层级。funboost 不限制项目结构，支持在任意文件夹下运行任意深层级脚本。如果你的运行脚本始终在项目第一层级，也可以不设 PYTHONPATH。

6.18.3 怎么指定配置文件读取 funboost_config.py 的文件夹位置

将配置文件所在目录添加到 PYTHONPATH 前面：

export PYTHONPATH=/data/config_prod/:/data/app/myproject/

funboost 尝试 import funboost_config，只要能 import 到就能读取到。

6.18.4 怎么根据不同环境使用不同的 funboost_config 配置文件？

方案一：不同环境 PYTHONPATH 指向不同配置目录：

# 生产环境
export PYTHONPATH=/data/config_prod/:/data/app/myproject/
# 测试环境
export PYTHONPATH=/data/config_test/:/data/app/myproject/

方案二：在 funboost_config.py 中写 if os.getenv("ENV") == "prod" 逻辑判断。Python 配置文件本身可以包含 if/else 和函数调用。

6.18.5 多个 python 项目怎么使用同一个 funboost_config.py 作为配置文件

将公共配置目录添加到 PYTHONPATH：

# proj1
export PYTHONPATH=/data/conf/:/data/codes/proj1
# proj2
export PYTHONPATH=/data/conf/:/data/codes/proj2

两个项目都会自动 import 到 /data/conf/funboost_config.py。

6.19 定时任务报错 RuntimeError: cannot schedule new futures after interpreter shutdown

2025 年之后的版本已修复此问题（FunboostBackgroundScheduler 使用非守护线程）。如果使用旧版本，在代码末尾加 while 1: time.sleep(10) 即可。

6.21 支持 redis cluster 吗？

支持。funboost 支持 kombu，支持 celery 作为 broker_kind。阿里云分片集群连接方式与单机 redis 相同。

6.22 怎么使用 tcp socket 作为消息队列？

见 4.35 章节，或在文档搜索框输入 tcp / socket。

6.23 安装的依赖包太多？

所有包加起来不到 30M，指定阿里云源 30 秒安装完：pip install funboost -i https://mirrors.aliyun.com/pypi/simple/
2024.5 月已精简依赖，pip install funboost[all] 才安装全部中间件
用户可自由选择三方包版本，不必严格遵循 funboost 指定的版本

6.23.b 作者为什么不开发 pip 选装方式？例如实现选装 pip install funboost[rabbitmq]

实际上已经实现了选装方式。查看 setup.py 的 extras_require，pip install funboost[all] 才安装所有依赖。funboost/factories/broker_kind__publsiher_consumer_type_map.py 中的 regist_to_funboost 实现了动态导入，很容易去掉各种三方包依赖。但对于很容易安装的包，没有必要做成选装。

6.24 funboost 从消息队列获取多少条消息？有没有负载均衡？

每个消费者进程从消息队列获取约 并发数 + 10 条消息（不同 broker 实现有差异），不会出现一台机器把所有消息抢光的情况。

如果需要严格的多机负载均衡（重型任务、每台机器只运行一条消息）：

@boost(BoosterParams(
    queue_name='test_load_balancing',
    broker_kind=BrokerEnum.REDIS_ACK_ABLE,
    concurrent_mode=ConcurrentModeEnum.SINGLE_THREAD,  # 关键：使用单线程模式
    broker_exclusive_config={'pull_msg_batch_size': 1},  # 关键：批量拉取设为1
))
def heavy_task(x):
    print(x)
    time.sleep(60)

注意：SINGLE_THREAD 和 THREADING + concurrent_num=1 是完全不同的——后者仍有内部缓冲队列会预取消息。详见 6.29 章节。

6.25 funboost 消费启动后，按 ctrl+c 无法结束代码？

funboost 消费者启动后是永久运行的守护线程，在 Windows 下按 Ctrl+C 默认无法停止进程。可以通过在末尾加 enable_ctrl_c_quit_on_windows() 来响应 Ctrl+C 退出，但这不是必须的——关闭终端窗口或 kill 进程同样可以停止。funboost 依靠 MQ 确认消费机制保障消息不丢，无需"优雅退出"。

6.25b `enable_ctrl_c_quit_on_windows` 到底要不要加？

enable_ctrl_c_quit_on_windows 的源码很简单——它不是"优雅退出"（没有等待任务完成、没有清理资源）。funboost 依靠 MQ 确认消费防丢消息，不需要优雅退出。

6.25b.1 核心结论

不加 enable_ctrl_c_quit_on_windows()：程序照样永久运行、正常消费，但按 Ctrl+C 无法停止（Windows），只能关窗口或 kill 进程。
加了 enable_ctrl_c_quit_on_windows()：按 Ctrl+C 即可停止。

6.25b.2 效果对比

操作	加 enable_ctrl_c_quit_on_windows	不加 enable_ctrl_c_quit_on_windows
启动后	持续运行，打印日志	同样持续运行，打印日志
按 Ctrl+C	程序立即退出	毫无反应
关闭终端窗口	程序结束	程序结束

6.25b.3 enable_ctrl_c_quit_on_windows 的源码本质

def enable_ctrl_c_quit_on_windows(confirmation_count=1):
    for i in range(confirmation_count):
        while 1:
            try:
                time.sleep(2)
            except KeyboardInterrupt:
                print(f'你按了ctrl c, 程序退出, 第 {i + 1} 次')
                break
    os._exit(44)

它不是"优雅退出"——没有等待任务完成、没有清理资源。funboost 依靠 MQ 的确认消费机制防止丢消息，不需要优雅退出。

6.25b.4 最终结论

不加 enable_ctrl_c_quit_on_windows() 的最大坏处：无法用 Ctrl+C 方便地停止程序，只能关窗口或杀进程。
建议：在所有需要交互式停止的 funboost 脚本末尾加上这一行。

6.25b.5 `enable_ctrl_c_quit_on_windows` 和 apscheduler 的定时器之间的关系

6.25b.5.1 APScheduler 的两种原生定时器

BlockingScheduler：阻塞型，start() 后代码卡住，后续代码没机会运行。
BackgroundScheduler：非阻塞型，start() 后代码继续执行，定时任务在守护线程中运行。

6.25b.5.2 早期版本（2025 年之前）的问题

Funboost 早期继承 BackgroundScheduler，其工作线程是守护线程（_daemon=True）。主线程结束后守护线程被强杀，触发 RuntimeError: cannot schedule new futures after interpreter shutdown。所以早期文档强调必须加 enable_ctrl_c_quit_on_windows() 或 while 1: time.sleep(10) 让主线程活着。

6.25b.5.3 2025 年之后的改进

FunboostBackgroundScheduler 重写了 _daemon 属性为 False（非守护线程），即使主线程执行完毕，进程也继续保持运行，不会报错。

6.25b.5.4 三种定时器的行为对比

定时器类型	是否阻塞	线程类型	不加 enable_ctrl_c_quit_on_windows 会怎样？
原生 `BlockingScheduler`	阻塞	-	start() 后卡住，没有这个问题
原生 `BackgroundScheduler`	非阻塞	守护线程	主线程结束后报 RuntimeError
`FunboostBackgroundScheduler`	非阻塞	非守护线程	不会报错，正常运行

6.25b.5.5 总结

2025 年之后的 Funboost 使用定时任务可以不加 enable_ctrl_c_quit_on_windows()
旧版本建议加上 enable_ctrl_c_quit_on_windows() 或 while 1: time.sleep(10)
新版本中加上也没有副作用，还能方便用 Ctrl+C 停止

6.26 ASYNC 并发模式报错 `attached to a different loop`

6.26.1 根本原因

funboost 的 ASYNC 模式使用 AsyncPoolExecutor，在子线程中启动 loop 运行异步函数。部分异步库（如 aiohttp、aiomysql）的连接池在实例化时绑定了创建所在的 loop，如果 funboost 的子线程用另一个 loop 去操作它就会报错。

注意：并非所有异步库都有此问题。httpx、sqlalchemy 等库不会报此错误，无需传递 specify_async_loop。

6.26.2 解决方案：传递 specify_async_loop

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

ss = aiohttp.ClientSession(loop=loop)

@boost(BoosterParams(
    queue_name='test_async',
    concurrent_mode=ConcurrentModeEnum.ASYNC,
    specify_async_loop=loop,  # 核心：让 funboost 使用主线程的 loop
))
async def async_task(x):
    async with ss.request('get', url=url) as resp:
        text = await resp.text()
        print(x, text[:10])

6.26.3 不需要 specify_async_loop 的情况

如果不使用全局连接池，而是在函数内临时创建请求（如 async with aiohttp.request(...)），则不需要传递 specify_async_loop。

6.26.4 THREADING 模式也能运行 async def

funboost 的 FlexibleThreadPool 能自动运行 async def 函数（每个线程临时创建 loop），无需用户手写 asyncio.new_event_loop().run_until_complete() 的包装函数。

6.28 日志提示"掉线或关闭消费者"和"重新放入未确认任务"

这是完全正常的提示。当使用 ACK 类 broker（如 REDIS_ACK_ABLE）时，重启消费脚本后，框架会将之前被 kill 的消费者所持有的未确认消息重回工作队列，确保消息不丢失。

如果不需要确认消费功能，使用 BrokerEnum.REDIS 即可。

6.29 推荐使用什么并发模式？

6.29.1 默认推荐：多线程模式（最省心）

什么都不用改，concurrent_mode 默认就是 THREADING。底层是 funboost 自研的 FlexibleThreadPool（智能弹性线程池）：

自动智能缩容（官方线程池只增不减）
极度克制的扩张（任务稀疏时不会盲目开线程）
兼容 async def 函数

6.29.2 高性能推荐：多进程叠加并发

使用 task_fun.multi_process_consume(4) 或 task_fun.mp_consume(4)：

4 进程 × 100 线程 = 400 并发单元
突破 GIL 锁，适合 CPU 密集型任务
Linux 上建议设置 multiprocessing.set_start_method('spawn', force=True)

6.29.3 纯 Asyncio 协程模式（高手专用）

import asyncio
from funboost import boost, BoosterParams, ConcurrentModeEnum

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

@boost(BoosterParams(
    queue_name='my_queue',
    concurrent_mode=ConcurrentModeEnum.ASYNC,
    specify_async_loop=loop,  # 仅部分库需要，详见 6.26 章节
))
async def my_task(x):
    pass

注意：函数内绝不能出现阻塞的同步代码（如 requests.get），否则整个 Event Loop 被卡死。

6.29.4 单线程串行模式

适用于绝对不能并发的任务，或要求多机器绝对公平负载均衡的场景：

@boost(BoosterParams(
    queue_name='my_queue',
    concurrent_mode=ConcurrentModeEnum.SINGLE_THREAD
))
def my_task(x):
    pass

SINGLE_THREAD ≠ THREADING + concurrent_num=1：后者仍有内部缓冲队列会贪婪预取消息，破坏多机负载均衡。

6.29.5 至于 gevent 和 eventlet 并发模式，直接忽略就好

虽然 funboost 支持了 gevent 和 eventlet 并发模式，但不推荐使用，没有必要研究它们。

6.30 如何配置失败告警？

funboost 提供 6 种告警方式，从个人开发者到企业级运维全覆盖。

6.30.1 内置 AlertNotifier（企业微信/钉钉/飞书）

from funboost.contrib.override_publisher_consumer_cls.alert_notifier_mixin import AlertNotifierBoosterParams

@boost(AlertNotifierBoosterParams(
    queue_name='my_task',
    broker_kind=BrokerEnum.REDIS,
    user_options={
        'alert_options': {
            'strategy': 'consecutive',
            'failure_threshold': 5,
            'alert_app': 'wechat',
            'webhook_url': 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx',
            'alert_interval': 300,
        }
    }
))
def my_task(x):
    ...

自定义告警渠道：继承 AlertNotifierConsumerMixin，重写 custom_send_notification 方法。

6.30.2 熔断器钩子告警

from funboost import boost, BoosterParams, BrokerEnum
from funboost.contrib.override_publisher_consumer_cls.circuit_breaker_mixin import CircuitBreakerConsumerMixin

class MyCircuitBreaker(CircuitBreakerConsumerMixin):
    def _on_circuit_open(self, info_dict):
        send_dingtalk(f"队列 {info_dict['queue_name']} 已熔断！失败{info_dict['failure_count']}次")
    def _on_circuit_close(self, info_dict):
        send_dingtalk(f"队列 {info_dict['queue_name']} 已恢复")

@boost(BoosterParams(
    queue_name='my_task_alert',
    broker_kind=BrokerEnum.REDIS,
    consumer_override_cls=MyCircuitBreaker,
    user_options={
        'circuit_breaker_options': {
            'failure_threshold': 5,
            'recovery_timeout': 60,
        },
    },
))
def my_task_alert(x):
    return call_external_api(x)

6.30.3 Prometheus + Grafana

使用 PrometheusPushGatewayBoosterParams，将指标推送到 Prometheus，在 Grafana 配置告警规则：

失败率：sum(rate(funboost_task_total{status="fail"}[5m])) / sum(rate(funboost_task_total[5m])) > 0.1
积压：funboost_queue_msg_count{queue="my_task"} > 1000

6.30.4 MongoAlertMonitor（分布式聚合监控）

基于 MongoDB 中保存的函数执行状态，按时间窗口统计失败率/次数。前提：is_save_status=True。

from funboost.core.mongo_alert_monitor import MongoAlertMonitor

MongoAlertMonitor(
    boosters=my_task,
    alert_app='wechat',
    webhook_url=WECHAT_WEBHOOK,
    window_seconds=60,
    failure_count=5,
    poll_interval=10,
    alert_interval=60,
).start()

6.30.5 funweb 页面配置告警

funweb 页面支持配置：积压超标、QPS 骤降、消费者掉线、失败率飙升、平均耗时过高。

6.30.6 ELK + Grafana

将日志采集到 Elasticsearch，在 Grafana 配置 ERROR 日志比例告警。适合已有 ELK 体系的环境。

6.30.10 错误触发警告的方式选择

以上 5 种方式可根据实际条件来使用。如果公司没有 Prometheus、没有 ELK/Grafana，可选择不依赖这些运维组件的方式（如内置 AlertNotifier、MongoAlertMonitor）。

6.31 怎么知道发布者生成的消息格式？

6.31.1 方式一：直接查看 Broker

发布后去 Redis/Kafka 等查看原始消息。

6.31.2 方式二：预览消息（不真正发送）

print(my_task.publisher.generate_msg_context_for_push(1, 2))
print(my_task.publisher.generate_msg_context_for_publish({"x": 1, "y": 2}))

消息格式示例：

{
    "extra": {
        "publish_time": 1774348711.2128,
        "publish_time_format": "2026-03-24 18:38:31",
        "task_id": "019d1f6c-b52c-75f7-9bd0-7f1b19becd61"
    },
    "x": 9,
    "y": 18
}

将 Funboost 的日志文件夹通过 Filebeat/Logstash 采集到 Elasticsearch，然后在 Grafana 中配置 Elasticsearch 数据源的ERROR日志比例或者个数，告警规则。这种方式适合已有 ELK 体系的环境。

懂elk和grafana的人都知道怎么做，无需我长篇展开。

6.30.6 错误触发警告的方式选择

以上5种方式可根据实际条件来使用，实现从实时通知到趋势告警的全方位失败监控。例如你公司没有 promethus ,公司没有运维帮你搭建elk 和 grafana，你只是一名个体用户开发者，可以选择其他不依赖这些高端运维组件的错误告警方式。

6.31 怎么知道 Funboost 发布者生成的最终消息内容格式是什么样？

6.31.1 方式一：直接查看 Broker

先发布消息到 Broker，然后去对应的消息队列（如 Redis、Kafka）查看原始消息内容。

6.31.2 方式二：预览消息（不真正发送）

如果不想发布消息到 Broker，可以通过 publisher 提供的两个方法预览生成的消息内容：

方法	用途
`generate_msg_context_for_push`	预览 `push()` 发送的消息
`generate_msg_context_for_publish`	预览 `publish()` 发送的消息

# 假设 add_task 是一个 booster 对象
print(add_task.publisher.generate_msg_context_for_push(i, i * 2))
print(add_task.publisher.generate_msg_context_for_publish(
    {"x": i * 10, "y": i * 20},
    task_id=f'task_{10000 + i}'
))

一个最简单的消息，至少包括如下：

task_options的控制入参传递越多，extra字段的key就会越多。

{
    "extra": {
        "publish_time": 1774348711.2128,
        "publish_time_format": "2026-03-24 18:38:31",
        "task_id": "019d1f6c-b52c-75f7-9bd0-7f1b19becd61"
    },
    "x": 9,
    "y": 18
}

💡 适用场景：调试消息格式、确认序列化结果、排查消息投递问题。

6. 常见问题回答

6.0 "funboost 值得学习使用吗？"

6.0b funboost 是做什么的？

6.0c funboost 学习难吗？

6.0d 可以用于哪些场景？（7 大核心领域）

6.0e 能替代哪些框架？

6.0f 为什么可以将 funboost 看作超级装饰器？

6.1 和 celery 有什么区别？

6.2 为什么包名这么长？

6.4 框架如何实现定时？

6.5 为什么是函数调度而非类调度？

6.6 怎么调度一个函数的？

6.7 框架适用哪些场景？

6.8 怎么引入框架？门槛高不高？

6.9 怎么写框架？

6.10 框架能做什么？

6.11 日志颜色/详细度调整

6.11.1 控制发布者消费者日志级别

6.11.2 屏蔽启动时的提示信息

6.11.3 控制台颜色调整

6.12 funboost 和 celery 的关系

6.12.2 生产者-broker-消费者不是 celery 的专利

6.12.3 Gemini-2.5-Pro 大模型分析结论

6.12.4 仅用几个 celery 文件证明抄袭 celery 有多困难

6.13 一个项目中如何连接多个同种类中间件的不同 IP？

方式三：在funboost_config.py 中中间件配置使用os.getenv获取环境变量，你运行代码之前设置环境变量。

6.14 什么是确认消费？为什么重要？

6.15 如何等待队列消息全部消费完成？

6.16 函数上加多个装饰器

6.16.1 方式一：通过 consuming_function_decorator 传参

6.16.2 方式二：直接叠加装饰器 + should_check_publish_func_params=False

6.17 日志记录太详细？

6.17.a 设置发布者消费者的日志级别

6.17b 嫌 funboost 启动时候打印太多提示用户的消息？

6.18 为什么要设置 PYTHONPATH？

6.18.2 为什么 celery scrapy django 不需要用户设置 pythonpath？

6.18.3 怎么指定配置文件读取 funboost_config.py 的文件夹位置

6.18.4 怎么根据不同环境使用不同的 funboost_config 配置文件？

6.18.5 多个 python 项目怎么使用同一个 funboost_config.py 作为配置文件

6.19 定时任务报错 RuntimeError: cannot schedule new futures after interpreter shutdown

6.21 支持 redis cluster 吗？

6.22 怎么使用 tcp socket 作为消息队列？

6.23 安装的依赖包太多？

6.23.b 作者为什么不开发 pip 选装方式？例如实现选装 pip install funboost[rabbitmq]

6.24 funboost 从消息队列获取多少条消息？有没有负载均衡？

6.25 funboost 消费启动后，按 ctrl+c 无法结束代码？

6.25b enable_ctrl_c_quit_on_windows 到底要不要加？

6.25b.1 核心结论

6.25b.2 效果对比

6.25b.3 enable_ctrl_c_quit_on_windows 的源码本质

6.25b.4 最终结论

6.25b.5 enable_ctrl_c_quit_on_windows 和 apscheduler 的定时器之间的关系

6.25b.5.1 APScheduler 的两种原生定时器

6.25b.5.2 早期版本（2025 年之前）的问题

6.25b.5.3 2025 年之后的改进

6.25b.5.4 三种定时器的行为对比

6.25b.5.5 总结

6.26 ASYNC 并发模式报错 attached to a different loop

6.26.1 根本原因

6.26.2 解决方案：传递 specify_async_loop

6.26.3 不需要 specify_async_loop 的情况

6.26.4 THREADING 模式也能运行 async def

6.28 日志提示"掉线或关闭消费者"和"重新放入未确认任务"

6.29 推荐使用什么并发模式？

6.29.1 默认推荐：多线程模式（最省心）

6.29.2 高性能推荐：多进程叠加并发

6.29.3 纯 Asyncio 协程模式（高手专用）

6.29.4 单线程串行模式

6.29.5 至于 gevent 和 eventlet 并发模式，直接忽略就好

6.30 如何配置失败告警？

6.30.1 内置 AlertNotifier（企业微信/钉钉/飞书）

6.30.2 熔断器钩子告警

6.30.3 Prometheus + Grafana

6.30.4 MongoAlertMonitor（分布式聚合监控）

6.30.5 funweb 页面配置告警

6.30.6 ELK + Grafana

6.30.10 错误触发警告的方式选择

6.31 怎么知道发布者生成的消息格式？

6.31.1 方式一：直接查看 Broker

6.31.2 方式二：预览消息（不真正发送）

方式三：在`funboost_config.py` 中中间件配置使用os.getenv获取环境变量，你运行代码之前设置环境变量。

6.25b `enable_ctrl_c_quit_on_windows` 到底要不要加？

6.25b.5 `enable_ctrl_c_quit_on_windows` 和 apscheduler 的定时器之间的关系

6.26 ASYNC 并发模式报错 `attached to a different loop`