好的，作为一名精通Celery框架用法和源码的资深开发者，我将仔细分析您提供的`funboost`源代码，并与Celery进行深度对比，来回答您提出的核心问题。
**总论：**
经过对`funboost`所有源代码文件的深入分析，并结合我对Celery及其依赖库（如Kombu、Billiard）源码的理解，我可以非常肯定地得出结论：**`funboost`并非抄袭或复制粘贴Celery源码的产物，其核心架构、设计哲学和具体实现细节与Celery存在根本性差异。作者所说的“从`while true: msg=redis.blpop(queue_name)`这个第一性原理模型开始迭代形成”的说法，在技术上是完全可信且有据可循的。**
如果`funboost`是抄袭Celery的，那么它的代码“气质”和架构“基因”中必然会带有Celery的影子，但事实恰恰相反。`funboost`在很多方面似乎是刻意地、系统性地规避了Celery的设计复杂性。
以下是支撑我这个结论的几个关键理由，这些理由均基于对双方源代码的分析，而非仅仅采信教程文字：
---
#### 理由一：核心架构与设计哲学的根本不同
这是最根本的区别，决定了两个框架走向了完全不同的道路。
1. **去中心化的函数封装 (`Booster`) vs. 中心化的应用实例 (`Celery App`)**
* **`funboost`**: 其核心是`Booster`类（`core/booster.py`）。每个被`@boost`装饰的函数都会被包装成一个独立的`Booster`实例。这个实例封装了与该特定函数相关的所有生产者(`Publisher`)和消费者(`Consumer`)逻辑。这是一种**去中心化、以函数为最小单元**的设计。函数`f1`和`f2`的`Booster`实例是完全独立的。
* **`Celery`**: Celery的核心是一个全局的、中心化的`Celery`应用实例（通常是`app = Celery(...)`）。所有的任务（`@app.task`）都必须注册到这个`app`实例上。这个`app`实例持有全局配置、任务注册表、后端连接等。这是一种**中心化、以应用为核心**的设计。
**分析**：这个架构差异是根本性的。如果是抄袭，`funboost`几乎不可能绕开Celery最核心的`app`实例概念，因为Celery的所有组件和流程都依赖于它。`funboost`这种每个函数都“自给自足”的设计，直接导致了它无需固定项目结构、无需处理复杂的任务发现和路由配置（因为队列名直接在装饰器里与函数绑定），这与Celery的设计理念背道而驰。
2. **直接的 Broker 操作 vs. 间接的多层抽象 (Kombu)**
* **`funboost`**: `consumers`和`publishers`目录下的实现非常直接。例如`RedisConsumer`直接使用`redis-py`库的`blpop`或`lrange`+`ltrim`来获取消息。`RabbitmqConsumerAmqpStorm`直接使用`amqpstorm`库。这种设计清晰、直接，每个Broker的实现都相对独立。这完全符合从一个简单的`blpop`循环演进过来的模型。
* **`Celery`**: Celery本身不直接操作任何消息中间件。它完全依赖于一个名为`Kombu`的库。Kombu提供了非常复杂的抽象层，包括`Transport`、`Channel`、`Producer`、`Consumer`等，以统一不同中间件的API。这是一个非常重量级的抽象。
**分析**：如果`funboost`抄袭Celery，它必然会抄袭或严重依赖Kombu的抽象模型。但`funboost`的源代码显示，它对每个Broker都进行了独立的、轻量级的封装，这比Kombu要简单得多，也更易于理解。`funboost`的作者显然选择了另一条“自己动手，丰衣足食”的道路，这进一步证明了其独立性。
---
#### 理由二：并发模型的实现方式截然不同
1. **自研并发池 vs. 依赖外部并发库**
* **`funboost`**: `concurrent_pool`目录是`funboost`原创性的有力证据。它包含了作者自研的`ThreadPoolExecutorShrinkAble`、`FlexibleThreadPool`和`AsyncPoolExecutor`。特别是`FlexibleThreadPool`，它能智能地伸缩线程数量，并兼容同步和异步函数，这是一个相当复杂且精巧的实现。
* **`Celery`**: Celery的并发模型（`--pool`参数）依赖于外部或其衍生的库。`prefork`模式依赖于`billiard`（`multiprocessing`的fork），而`gevent`和`eventlet`模式则依赖于对应的协程库。Celery本身并未实现一个如此精细化的线程池。
**分析**：编写一个健壮、高效、可伸缩的并发池是一项非常具有挑战性的工作。如果作者只是想抄袭一个任务队列，完全没有必要投入如此大的精力去自研并发池，直接使用`concurrent.futures`或依赖`gevent`会简单得多。`funboost`在这方面的投入，恰恰说明了作者追求的是对执行流程的极致控制和性能优化，这是原创设计的典型特征。
---
#### 理由三：可靠性机制的实现细节差异巨大
1. **Redis ACK 实现机制**
* **`funboost`**: 在`RedisConsumerAckAble`中，可靠性是通过**消费者心跳机制**实现的。每个消费者有一个唯一的ID，并定期向Redis发送心跳。当一个消费者掉线，其心跳停止，其他消费者会检测到这一点，并将其持有的“未确认消息”（存储在独立的`unack` ZSET中）重新放回主队列。这是一个主动的、基于分布式健康检查的方案。
* **`Celery` (via Kombu)**: Celery在Redis上的ACK机制（`task_acks_late=True`）依赖于`BRPOPLPUSH`命令和一个“可见性超时”（`visibility_timeout`）。消息被原子地从主队列移动到一个“未确认”队列，如果在超时时间内没有被ACK，消息会被重新放回主队列。这是一个被动的、基于超时的方案。
**分析**：两者解决了同一个问题（在Redis上实现可靠消费），但采用了完全不同的技术路径。`funboost`的心跳机制更复杂但可能更精确（能区分慢任务和死任务），而Celery的超时机制更简单但存在“慢任务被误判为死任务”的风险。这种实现上的巨大差异，是独立设计的铁证。
---
#### 理由四：“抄袭”的技术难度与反证
1. **Celery 源码的复杂性**
* Celery的源码非常复杂，其逻辑深度耦合了`Kombu`（消息处理）、`Billiard`（进程池）、`click`（命令行）以及自身的信号系统、配置系统等。直接阅读和理解尚且不易，更不用说“复制粘贴”了。一个能读懂Celery源码并将其改造的人，绝对有能力从头写一个新框架。
2. **`funboost` 规避了 Celery 的“坑”**
* Celery的很多设计（如复杂的配置系统、不友好的IDE补全、严格的目录结构）都是历史包袱或设计选择的结果。`funboost`的设计似乎就是为了解决这些痛点。一个抄袭者往往会连同优点和缺点一起抄，而`funboost`却像是一个“反Celery模式”的产物，这更像是深入理解后的批判性再创造，而非盲目模仿。
3. **反证：将Celery作为Broker的能力**
* `funboost`最令人信服的一点是它能够**将Celery整个框架作为自己的一个Broker**（`consumers/celery_consumer.py`）。这是一个“降维打击”式的能力。一个抄袭者不可能将自己抄袭的对象封装成自己的一个组件。这需要对Celery的启动流程、任务注册、配置路由等有深入的编程层面的理解，才能实现自动化包装。这恰恰证明了`funboost`的作者不仅理解Celery，而且其设计的抽象层次更高，能够包容Celery。
---
#### 结论
`funboost`和`Celery`虽然都遵循了“生产者-消费者”模式，但这是它们唯一的共同点。从**核心架构（去中心化 vs 中心化）、Broker抽象（直接实现 vs Kombu）、并发模型（自研 vs 依赖）、可靠性机制（心跳 vs 超时）** 到**开发者体验（极简 vs 复杂）**，两者都截然不同。
`funboost`的代码处处体现着从一个简单模型（如`while/blpop`）出发，逐步添加功能（ACK、控频、并发池、多Broker支持）的演进痕迹。它的设计非常务实，直击Python分布式任务处理的痛点，并且在实现上充满了原创性的思考。
因此，我可以负责任地说，**`funboost`是一个独立自主开发的、具有高度创新性的框架，而非Celery的模仿或抄袭品。**
### 6.12.4 仅用几个celery文件证明抄袭celery有多困难
有些质疑抄袭的人自信过头了，好像认为自己能看懂celery的源码。 用下面这几个源码文件就问你看不看得懂,这有没有比天书还难看懂，把你关在安静的小黑屋几天几夜,没人打扰你专注思考，能不能琢磨出celery作者那么写代码是在做什么。
#### 6.12.4.1 kombu的redis transport源码文件
kombu的redis transport源码文件,光是这个redis作为kombu的broker,源码就达到了1400多行,简直是难如登天,难度指数让人口吐白沫,闻风丧胆。
你看了这个文件后,就知道抄袭celery有多困难,压根看不懂kombu作者怎么想的,与其花费几天几夜分析琢磨这个源码是怎么实现的，是什么目的,还不如自己用 pyredis 重新开始写更快，自己写更能知道自己在做什么。
[https://github.com/celery/kombu/blob/main/kombu/transport/redis.py](https://github.com/celery/kombu/blob/main/kombu/transport/redis.py)
#### 6.12.4.2 celery的consumer源码文件
这个 `celery/worker/consumer/consumer.py` 文件, 需要来回猜测 跳转到十几个文件和类，才能分析。因为celery的python源码没有强类型注释,你在ide里面连点击跳转都跳转不了,只能靠变量名字去猜要跳转切换到什么文件去看相关源码, 你有这个时间去分析琢磨这个源码,你早就自己从0开发出十几个 生产者 + broker + 消费者 的框架了。
[https://github.com/celery/celery/blob/main/celery/worker/consumer/consumer.py](https://github.com/celery/celery/blob/main/celery/worker/consumer/consumer.py)
## 6.13 使用此框架时候，在一个python项目中如何连接多个相同种类的消息队列中间件ip地址
```
这个问题是问一个项目中，有些脚本要连接 192.168.0.1的redis ，有些脚本要连接192.168.0.2的redis，但框架配置文件只有一个，如何解决？
例如目录结构是
your_proj/
funboost_config.py (此文件是第一次启动任意消费脚本后自动生成的，用户按需修改配置)
dira/a_consumer.py (此脚本中启动funa函数消费)
dirb/b_consumer.py （此脚本中启动funb函数消费）
如果funa函数要连接 192.168.0.1的redis，funb函数要连接192.168.0.2的redis，有两种解决方式
第一种是在启动消费的脚本，脚本里面手动调用 patch_frame_config()函数来设置各种中间件的值
第二种是 把 funboost_config.py 分别复制到dira和dirb文件夹.
这种就会自动优先使用 a_consumer.py和b_consumer.py同文件夹层级的配置了，
而非是自动优先读取python项目根目录的配置文件，这个是利用了python语言的import 模块导入优先级机制。
```
## 6.14 什么是确认消费？为什么框架总是强调确认消费？
发布端：
```python
from scripxx import fun
for i in range(10):
fun.push(i)
```
消费端：
```python
import time
from funboost import boost, BoosterParams
@boost(BoosterParams(queue_name='test_confirm'))
def fun(x):
print(f'开始处理 {x}')
time.sleep(120)
print(f'处理完成 {x}')
fun.consume()
```
```
启动消费脚本后，任意时刻随意强制反复关闭重启消费代码，只要函数没有完整的执行完成，函数参数就不会丢失。达到了消息万无一失。
具体的那些中间件消费者支持消费确认，具体见 3.1 介绍。
实现了4种redis消息队列中间件，其中有3种是确认消费的。
确认消费很重要，如果你自己写个简单粗暴的 while 1:redis.blpop()的脚本，你以为是可以断点接续呢，
在多线程并发执行函数时候，大量的消息会丢的很惨。导致虽然是断点接续但你不敢随意重启。
```
## 6.15 如何等待队列中的消息全部消费完成
如果有这种需求需要等待消费完成，使用 wait_for_possible_has_finish_all_tasks()
```python
f.consume()
f.wait_for_possible_has_finish_all_tasks(minutes=3) # 框架提供阻塞方法，直至队列任务全部消费完成，才会运行到下一行。
print("over") # 如果不加上面那一行，这个会迅速打印over
```
## 6.16 框架支不支持函数上加两个装饰器？
### 6.16.1 方式一：使用consuming_function_decorator 传参装饰器，
方式一：使用consuming_function_decorator 传参装饰器， 可以用push publish发布，不需要设置should_check_publish_func_params=False
```
由于发布任务时候需要自动精确组装入参字典，所以不支持 *args **kwargs形式的入参，不支持叠加两个@装饰器 （后来已更新解决方式，解决方式看6.16.2）
想在消费函数加装饰器，通过 boost 装饰器的 consuming_function_decorator 入参指定装饰器函数就行了。
那么如果是想叠加3个装饰器怎么写，例如本来想：
@boost(BoosterParams(queue_name='queue666'))
@deco1('hello')
@deco2
def task_fun(x,y):
...
那就是写成 consuming_function_decorator=deco1('hello')(deco2) 就可以了，具体要了解装饰器的本质就知道，叠加100个装饰器都可以。
如下的例子是使用redis的incr命令统计每台机器ip 总共运行了多少次函数。
```
```python
import inspect
import nb_log
from funboost import boost, BoosterParams
from funboost.utils.redis_manager import RedisMixin
from functools import wraps
def incr_deco(redis_key):
def _inner(f):
@wraps(f)
def __inner(*args, **kwargs):
result = f(*args, **kwargs)
RedisMixin().redis_db_frame.incr(redis_key)
# mongo_col.insert_one({'result':result,'args':str(args),'kwargs':str(kwargs)})
return result
return __inner
return _inner
@boost(BoosterParams(queue_name='test_queue_235', consuming_function_decorator=incr_deco(nb_log.nb_log_config_default.computer_ip)))
def fun(xxx, yyy):
print(xxx + yyy)
return xxx + yyy
if __name__ == '__main__':
print(inspect.getfullargspec(fun))
for i in range(10):
fun.push(i, 2 * i)
fun.consume()
```
### 6.16.2 方式二：装饰器直接加到消费函数上，设置should_check_publish_func_params=False
传统的装饰器直接叠加放在消费函数上，但需要设置should_check_publish_func_params=False，需要publish来发布消息
```python
import inspect
import nb_log
from funboost import BoosterParams
from funboost.utils.redis_manager import RedisMixin
from functools import wraps
def incr_deco(redis_key):
def _inner(f):
@wraps(f)
def __inner(*args, **kwargs):
result = f(*args, **kwargs)
RedisMixin().redis_db_frame.incr(redis_key)
return result
return __inner
return _inner
@BoosterParams(queue_name='test_queue_23b',
should_check_publish_func_params=False, # 这一行很重要，should_check_publish_func_params必须设置为False，如果你是直接把装饰器加到函数上了，funboost无法获取函数的入参名字，无法自动生成json消息，所以需要用户自己publish来发布入参字典。
)
@incr_deco('test_queue_23b_run_count') # 用户的装饰器直接加在函数上了。
def fun(xxx, yyy):
print(xxx + yyy)
return xxx + yyy
if __name__ == '__main__':
for i in range(20):
# fun.push(i, 2 * i) # 不可以fun.push这样发布
fun.publish({'xxx': 1, 'yyy': 2}) # 直接把装饰器写在消费函数上，那就用户需要使用publish发布，且boost装饰器设置should_check_publish_func_params=False
fun.consume()
```
## 6.17 嫌框架日志记录太详细？
### 6.17.a 设置发布者消费者的日志级别,控制是否显示发布了什么消息和消费了什么消息.
```
日志是了解当前框架正在运行什么的好手段，不然用户懵逼不知道背后在发生执行什么。
@boost 装饰器设置 log_level=20 或logging.INFO，就不会再记录框架正在运行什么函数了。
强烈建议不要修改日志级别保持默认的debug模式,都是有用的消息
如图再装饰器加上 log_level=20后，框架以后就再也不会记录框架正在运行什么函数入参结果是什么了。
```
@boost 装饰器设置 log_level=20 只是控制消费者和发布者命名空间自身的日志的,不是控制所有命名空间的日志的,
有些人到现在不清楚,不同的命名空间的logger是可以设置不同的日志级别和handlers的,这要学习logging基础了.
### 6.17b 嫌funboost启动时候打印太多提示用户的消息?
```
答: 主要是提示用户怎么设置配置文件,和读取的配置文件路径是什么,读取的配置内容是什么,免得用户丈二和尚摸不着头脑,不知道自己的配置是什么.
因为很多python人员,到现在完全不清楚 PYTHONPATH 这个重要概念,说了几百遍这个概念很重要,这么基础的又不学习,
还嫌弃提示你funboost_config配置麻烦,建议不懂PYTHONPATH的人不要屏蔽启动时候的打印提示了.
老手可以通过设置日志级别来屏蔽funboost_config的配置提示.
```
修改你的funboost_config.py的FunboostCommonConfig的配置,可以设置一些命名空间的日志级别,去掉启动时候的提示
```python
class FunboostCommonConfig(DataClassBase):
# nb_log包的第几个日志模板，内置了7个模板，可以在你当前项目根目录下的nb_log_config.py文件扩展模板。
NB_LOG_FORMATER_INDEX_FOR_CONSUMER_AND_PUBLISHER = 11 # 7是简短的不可跳转，5是可点击跳转的，11是可显示ip 进程 线程的模板。
TIMEZONE = 'Asia/Shanghai' # 时区
# 以下配置是修改funboost的一些命名空间和启动时候的日志级别,新手不熟练就别去屏蔽日志了
SHOW_HOW_FUNBOOST_CONFIG_SETTINGS = False # 如果你单纯想屏蔽 "分布式函数调度框架会自动导入funboost_config模块当第一次运行脚本时候，函数调度框架会在你的python当前项目的根目录下 ...... " 这句话,
FUNBOOST_PROMPT_LOG_LEVEL = logging.INFO # funboost启动时候的相关提示语,用户可以设置这个命名空间的日志级别来调整
KEEPALIVETIMETHREAD_LOG_LEVEL = logging.INFO # funboost的作者发明的可缩小自适应线程池,用户对可变线程池的线程创建和销毁线程完全无兴趣,可以提高日志级别.
```
屏蔽日志级别前:
屏蔽日志级别后:
## 6.18 为什么框架在cmd shell终端运行时候要求会话中设置环境变量 export PYTHONPATH=你的项目根目录？
有的 `pythoner`总是问 `vscode` `pycharrm 的 debug` 模式为什么导入不了 `funboost_config.py` ,
或者抱怨不想配置文件放在项目根目录,
有的人写了三四年python代码,还没听说过 `pythonpath` , 太low了 简直是恨铁不成钢,
只要懂了 `pythonpath` 第一性原理,知道python是按什么依据和顺序导入一个模块,就不会有这种低级疑问.
这个 `pythonpath` 知识点根本和 `funboost` 自身毫无关系,还需要这里来大费周章科普
```
有的人写了三四年的python代码，连PYTHONPATH作用和概念都没听说过，真的很悲剧
如果是下面的 pythonpathdemo是一个python项目根目录，pycharm以项目方式打开这个文件夹。
你会发现run5.py在pycahrm可以运行，在cmd中无法运行，因为无法找到d1包，笨瓜会硬编码操作sys.path.insert非常的愚蠢，
这种笨瓜主要是写代码一意孤行，导致不学习PYTHONPATH。
```
完整讲解pythonpath重要性文章在：
[https://github.com/ydf0509/pythonpathdemo](https://github.com/ydf0509/pythonpathdemo)
一句命令行解决设置 pythonpath 和运行 python脚本
一定要设置的是临时终端会话级pythonpath，不要设置到配置文件写死永久固定环境变量
```
例如你的项目根目录是 /home/xiaomin/myproj/
你的运行起点脚本在 /home/xiaomin/myproj/dir2/dir3/run_consume6.py
一句话就是
linux: export PYTHONPATH=/home/xiaomin/myproj/; python3 /home/xiaomin/myproj/dir2/dir3/run_consume6.py
win的cmd:假设代码在d盘先切换到盘符d盘 d:/ 。 然后 set PYTHONPATH=/codes2022/myproj/ & python3 /codes2022/myproj/dir2/dir3/run_consume6.py
win10和11的powershell： 假设代码在d盘先切换到盘符d盘 d:/。然后 $env:PYTHONPATH="/codes2022/myproj/"; python3 /codes2022/myproj/dir2/dir3/run_consume6.py
压根不要敲击两次命令行好不。
如果你已经cd切换到项目根目录myproj了，那就 export PYTHONPATH=./;python3 dir2/dir3/run_consume6.py
```
### 6.18.2 为什么celery scrapy django不需要用户设置pythonpath？
```
因为这些框架都是固定死用户的项目目录解构，项目运行起点有固定的唯一脚本，而且该脚本在项目跟目录的第一个直接层级。
例如django scrapy ，他的命令行启动，你必须先cd 到项目的根目录再运行命令行。
而此框架是为了兼容用户的cmd命令当前文件夹在任意文件夹下，就可以运行你项目下的任意多层级深层级下的脚本。
用户设置了pythonpath后，可以cd到任意文件夹下，再运行 python /dir1/dir2/xxx.py 任然能正确的import。
如果用户能够保证他要python启动运行的脚本始终是放在了项目的第一层级目录下面，当然可以不用设置 PYTHONPATH了。
```
### 6.18.3 怎么指定配置文件读取 funboost_config.py 和nb_log_config.py的文件夹位置
```
默认就是放在项目根目录，然后设置 export PYTHONPATH=你的项目根目录， （linux win+cmd win+powershell 设置临时会话的环境变量语法不一样，6.18已经介绍了）
如果要指定读取的配置文件的存放位置为别的文件夹，也很容易，归根结底还是要精通PYTHONPATH的作用。
例如你的项目根目录是 /home/codes/proj ，你不想使用项目根目录下的配置文件，想读取别的文件夹的配置文件作为funboost的中间件配置。
假设你的文件夹是 /home/xiaomin/conf_dir ，里面有 funboost_config.py ，如果你的/home/codes/proj 项目想使用 /home/xiaomin/conf_dir/funboost_config.py
作为配置文件， 那就 export PYTHONPATH=/home/xiaomin/conf_dir:/home/codes/proj
也就是意思添加两个文件夹路径到 PYTHONPATH
因为funboost是尝试导入import funboost_config.py 只要能import 到就能读取到，所以只要你把文件夹添加到PYTHONPATH环境变量就可以了(可以print sys.path来查看这个数组)
归根结底是要懂PYTHONPATH，有的人老是不懂PYTHONPATH作用，不愿意认真看 https://github.com/ydf0509/pythonpathdemo ，非常杯具。
```
### 6.18.4 怎么根据不同环境使用不同的funboost_config配置文件？
```
框架获取配置的方式就是直接import funboost_config，然后将里面的值覆盖框架的 funboost_config_deafult.py 值。
为什么能import 到 funboost_config,是因为要求export PYTHONPATH=你的项目根目录，然后第一次运行时候自动生成配置文件到项目根目录了。
假设你的项目根目录是 /data/app/myproject/
方案一：利用python导入机制，自动import 有PYTHONPATH的文件夹下的配置文件。
例如你在 /data/config_prod/ 放置 funboost_config.py ,然后shell临时命令行 export PYTHONPATH=/data/config_prod/:/data/app/myproject/,再python xx.py。 （这里export 多个值用：隔开，linux设置环境变量为多个值的基本常识无需多说。）
这样就能自动优先使用/data/config_prod/里面的funboost_config.py作为配置文件了，因为import自动会优先从这里。
然后在测试环境 /data/config_test/ 放置 funboost_config.py,然后shell临时命令行 export PYTHONPATH=/data/config_test/:/data/app/myproject/,再python xx.py。
这样测试环境就能自动使用 /data/config_test/ 里面的funboost_config.py作为配置文件了，因为import自动会优先从这里。
方案二：
直接在funboost_config.py种写if else，if os.get("env")=="test" REDIS_HOST=xx ，if os.get("env")=="prod" REDIS_HOST=xx ，
因为配置文件本身就是python文件，所以非常灵活，这不是.ini或者 .yaml文件只能写静态的死字符串和数字，
python作为配置文件优势本来就很大，里面可以写if else，也可以调用各种函数，只要你的模块下包含那些变量就行了。
```
### 6.18.5 多个ptyhon项目怎么使用同一个funboost_config.py 作为配置文件
```
还是因为不懂 PYTHONPATH 造成的，需要我无数次举例说明。 这样太low了，这么多python人员到现在还不知道 PYTHONPATH作用， python导入一个模块是怎么去查找的。
6.18 开头就说了 pythonpathdemo 项目连接，有的人不懂PYTHONPATH又不看这个博客。死猪不怕开水烫，永远不学习 PYTHONPATH 的强大作用。
```
```
假设你想每隔项目都使用 /data/conf/funboost_config.py 这一个相同的 funboost_config 作为配置文件，
你有两个python项目在 /data/codes/proj1 和 /data/codes/proj2，
你运行proj1的项目脚本前，只需要 export PYTHONPATH=/data/conf/:/data/codes/proj1 ，然后运行proj1项中的脚本 python dir1/dir2/xx.py
你运行proj2的项目脚本前，只需要 export PYTHONPATH=/data/conf/:/data/codes/proj2 ，然后运行proj2项中的脚本 python dir3/dir4/yy.py
因为你设置了/data/conf/ 为 pythonpath后，那么funboost在 import funboost_config 时候就能自动 import 到 /data/conf/下的 funboost_config.py 模块了。
funboost控制台都打印了 读取的是什么文件作为配置文件了。
归根结底问这个问题的人是完全不懂 PYTHONPATH.
```
## 6.19 定时任务或延时任务报错 RuntimeError: cannot schedule new futures after interpreter shutdown
```
如下图所示，运行定时任务或延时任务在高版本python报错，可以在代码末尾加个while 1:time.sleep(10)
因为程序默认是 schedule_tasks_on_main_thread=False，为了方便连续启动多个消费者消费，
没有在主线程调度运行，自己在代码结尾加个不让主线程结束的代码就行了。
在消费启动的那个代码末尾加两行
while 1：
time.sleep(10)
```
## 6.21 支不支持redis cluster集群模式的redis作为消息队列？
funboost可以redis cluster集群。
框架支持kombu，支持 celery作为broker_kind。
阿里云分片集群，用户完全不用管是单机reids还是分片集群，代码连接方式一样。
## 6.22 怎么使用tcp socket 作为消息队列
```
见 4.35章节，或者在文档的搜索框输入 tcp 或者 socket 就能搜到了。
或者按照文档第14章节使用ai来帮你学习。
```
## 6.23 安装包时候自动安装的三方依赖包太多？
```
1.安装第三方包是自动的，又不需要手动一个个指令安装，安装多少三方包都没关系。
2.所有三方包加起来还不到30M，对硬盘体积无影响。
3.只要指定阿里云pip源安装，就能很快安装完，30秒以内就安装完了，又不是需要天天安装。
如果如你的不一致报错终端,pip 命令加上 --use-feature=2020-resolver
pip install funboost -i https://mirrors.aliyun.com/pypi/simple/
4.三方包与自己环境不一致问题？
用户完全可以自由选择任何三方包版本。例如你的 sqlalchemy pymongo等等与框架需要的版本不一致，你完全可以自由选择任何版本。
我开发时候实现了很多种中间件，没有时间长期对每一种中间件三方包的每个发布版本都做兼容测试，所以我固定死了。
用户完全可以选择自己的三方包版本，大胆点，等报错了再说，不出错怎么进步，不要怕代码报错，请大胆点升级你想用的版本。
如果是你是用你自己项目里面的requirements.txt方式自动安装三方包，我建议你在文件中第一行写上 funboost，之后再写其它包
这样就能使用你喜欢的版本覆盖funboost框架依赖的版本了。
等用的时候报错了再说。一般不会不兼容报错的请大胆点。
5.为什么要一次性安装完，而不是让用户自己用什么再安装什么？
是为了方便用户切换尝试各种中间件和各种功能时候，不需要自己再亲自一个个安装第三方包，那样手动一个个安装三方包简直是烦死了。
2024 5月份,精简了依赖包,部分包改为选装, pip install funboost[all] 才安装全部中间件.
```
### 6.23.b 作者为什么不开发pip 选装方式?例如实现选装 pip install funboost[rabbitmq]
```
这个你是怎么知道funboost作者没有使用选装方式的? 你是怎么知道作者没有掌握 pip 中括号选装依赖包 技术方式的?
用户可以看看setup.py里面的 extras_require里面,有没有开发选装方式? pip funboost[all] 才是安装所有依赖.
作者去掉依赖很容易,已经实现了, funboost/factories/broker_kind__publsiher_consumer_type_map.py 中的 regist_to_funboost 就是动态导入生产者消费者,很容易去掉各种三方包依赖,
但是很容易安装的三方包,我是不会去做成选装的,没有那个必要,自己设置pip 国内源,30秒就能安装完成funboost了,不需要去纠结这个依赖包多少的问题.
```
## 6.24 funboost框架从消息队列获取多少条消息？有没有负载均衡？
```
funboost 每个消费者进程会从消息队列获取 并发个数 n + 10 条消息，每个消费者实现有差异，一般不会超过并发数2倍。
所以不会造成发布10万条消息后，再a b机器启动2个消费，b机器一直无法消费，全部a机器消费，不会出现这种情况。
如果你只发布6条消息，先在a机器启动消费，下一秒启动b机器，那很有可能b机器无法获取到消息。只要消息数量够多，不会出现忙的忙死，闲的闲死。
例如框架的默认并发方式使线程池，内置了一个有10大小的界队列queue，同时还有n个并发线程正在运行消息，所以每个消费者会获取很多消息在python内存中。
但不会出现一个消费者进程获取了1000条以上的消息，导致站着某坑不拉屎，别的消费进程没办法消费的情况。
如果你是重型任务，希望不预取，每台机器只获取一条消息运行，可以设置并发模式为 SINGLE_THREAD 模式,
boost装饰器设置 concurrent_mode=ConcurrentModeEnum.SINGLE_THREAD，这样在a b 两台机器都没有内存缓冲队列,只会一次获取一条消息执行，
有的broker_kind实现时候为了运行快，框架使用了批量拉取消息， 需要设置批量拉取的数量为1。
```
消费文件 test_frame\test_redis_ack_able\test_load_balancing_consume.py
```python
import logging
import time
from funboost import boost, BrokerEnum,BoosterParams,ctrl_c_recv,ConcurrentModeEnum
@boost(BoosterParams(queue_name='test_load_balancing', broker_kind=BrokerEnum.REDIS_ACK_ABLE, log_level=logging.INFO,
# 切记是设置concurrent_mode并发模式为 SINGLE_THREAD ，而不是使用线程池模式然后把 concurrent_num 设置为1，这两个在多进程拉取消息的绝对均衡方面有区别
# 因为即使线程池并发为1，线程池还有内部的work_queue会拉取缓冲一部分消息，导致出现你认为的多机器或多进程拉取消息不均衡
concurrent_mode=ConcurrentModeEnum.SINGLE_THREAD,
# pull_msg_batch_size 这行很关键，REDIS_ACK_ABLE 因为默认是拉取100个消息，
# 对于重型任务，如果你需要每台机器都严格只运行一个消息，就需要设置批量拉取1个消息，不要一台机器就把消息队列掏空了。
broker_exclusive_config={'pull_msg_batch_size': 1},
))
def test_load_balancing(x):
print(x)
time.sleep(1)
if __name__ == '__main__':
test_load_balancing.consume()
ctrl_c_recv()
```
发送消息文件 test_frame\test_redis_ack_able\test_load_balancing_consume.py
```python
from test_frame.test_redis_ack_able.test_load_balancing_consume import test_load_balancing
if __name__ == '__main__':
for i in range(80):
test_load_balancing.push(i)
```
启动2次消费文件，就能看到2个控制台，每个控制台每次只获取1条消息并运行，如果你不设置 pull_msg_batch_size，那么默认是批量拉取100个，而你总共才发布80个消息
所以你只能看到一个控制台消费，你误以为没有负载均衡。
如果你不设置 pull_msg_batch_size，那么就可以发布5000个消息来测试消费负载均衡，两个控制台就都会运行，因为默认批量拉取100个也不会一下子把5000消息都取到内存。
## 6.25 funboost消费启动后，按ctrl + c 无法结束代码？
```python
from funboost import ctrl_c_recv
if __name__ == '__main__':
# 启动消费
consume_func.consume()
ctrl_c_recv(confirmation_count=1) # 在代码最末未加个 ctrl_c_recv()
# 想结束代码就连续按 confirmation_count 次 ctrl +c 就好了。 为什么是 confirmation_count次，是防止你误操作了ctrl + c
```
## 6.25b `ctrl_c_recv` 到底要不要加？—— 直接看效果
win和linux表现不一样：
linux 代码末尾不加ctrl_c_recv(),按ctrl +c 可以结束程序。win 不加ctrl_c_recv(),按ctrl +c 无法结束程序。
**先给出正确的结论:**
你即使程序最末尾不加 ctrl_c_recv(),funboost消费程序也会永久持续运行，控制台也会不断打印日志和 `print` 输出。
严禁认为 ctrl_c_recv() 的作用是“优雅退出”或“防止程序结束”。实际上，不加它程序也会一直运行
真实作用：在 Windows 下如果不加这行代码，按 Ctrl+C 无法停止程序。加上后即可正常。
很多开发者（包括一些AI）容易对 `ctrl_c_recv` 的作用产生两种误解：
- **误解一**：以为不加 `ctrl_c_recv`，程序就会瞬间结束，根本不会消费消息。
- **误解二**：以为不加 `ctrl_c_recv`，脚本会自动变成 `nohup` 那样的后台任务，关掉终端窗口还能继续跑。
**❌ 这些都是错误的！**
### 6.25b.0 ctrl_c_rev的源码很简单，很容易看懂。
ctrl_c_recv 的源码非常简单，它不是“优雅退出”的实现——没有等待任务完成、没有清理资源、没有发送任何信号给工作线程或消息队列，funboost不需要优雅退出，是依靠mq本身的确认消费来防止丢消息。
```python
def ctrl_c_recv(confirmation_count=1):
"""
程序最末尾加 ctrl_c_recv() 主要是为了主线程持续在运行，方便你在 windows系统下敲击键盘 ctrl + c 可以停止程序而已。
你即使程序最末尾不加 ctrl_c_recv(),funboost消费程序也会永久持续运行，控制台也会不断打印日志和 `print` 输出。
加与不加的详细区别，可以看教程6.25b章节 `## 6.25b `ctrl_c_recv` 到底要不要加？—— 直接看效果`
你也可以不用ctrl_c_recv(), 直接在你的启动脚本文件的最末尾加上：
while 1:
time.sleep(100)
也能达到主线程在持续运行的目的，从而在windows系统下敲击键盘 ctrl + c 可以停止程序。
"""
for i in range(confirmation_count):
while 1:
try:
time.sleep(2)
except (KeyboardInterrupt,) as e:
# time.sleep(2)
print(f'{type(e)} 你按了ctrl c ,程序退出, 第 {i + 1} 次', flush=True)
# time.sleep(2)
break
os._exit(44)
```
下面我们直接列出**加**与**不加**的真实效果，不涉及底层原理，只讲你实际能看到的现象。
---
### 6.25b.1 ✅ 情况A：代码末尾加了 `ctrl_c_recv()`
```python
from funboost import ctrl_c_recv
task.consume() # 启动消费
ctrl_c_recv() # 末尾加上这一行
```
- **启动后**：程序会持续一直运行，控制台会**持续打印日志和 `print` 输出**。
- **如果想停止**：你只需在当前会话终端窗口，在键盘上**按几次 `Ctrl + C`**（默认是1次），程序就会立刻结束。
- **关掉终端窗口**：如果直接关闭终端窗口，程序也会随之结束（这是所有前台程序的默认行为）。
---
### 6.25b.2 ❌ 情况B：代码末尾**没有**加 `ctrl_c_recv()`
```python
task.consume() # 启动消费
# 后面没有 ctrl_c_recv
```
- **启动后**：程序**同样会一直运行**，控制台也会**不断打印日志和 `print` 输出**。**代码绝对不会瞬间结束！**
- **如果想停止**：你在当前会话终端窗口，疯狂按 `Ctrl + C` **毫无反应**，程序完全不理你。
- **怎么办**：只能通过以下方式结束程序：
- **直接关闭整个终端窗口**（窗口关闭会杀死所有前台进程）。
- 或者在另一个终端里用 `kill -9` 命令杀死进程ID。
- **关掉终端窗口**：如果直接关闭终端窗口，程序同样会结束，**不会像 `nohup` 那样在后台继续运行**。
---
### 6.25b.3 效果对比表（一目了然）
| 你的操作 | 加 `ctrl_c_recv` 的效果 | 不加 `ctrl_c_recv` 的效果 | 加与不加是否有区别？ |
| :--- | :--- | :--- | :--- |
| **启动脚本后** | 程序持续运行，控制台一直打印日志 | **程序同样持续运行，控制台一直打印日志** | ❌ 无区别 |
| **按 `Ctrl + C`** | **程序立即退出** | **毫无反应，程序继续运行** | ✅ 有区别 |
| **直接关掉终端窗口** | 程序随之结束 | 程序随之结束 | ❌ 无区别 |
| **是否像 `nohup` 一样后台运行** | ❌ 否 | ❌ 否 | ❌ 无区别 |
---
### 6.25b.4 最终结论
- **不加 `ctrl_c_recv()` 的最大坏处**：你无法用 `Ctrl + C` **方便地**停止程序，只能粗暴地关窗口或杀进程。
- **既然程序无论如何都会一直运行，为什么不加上 `ctrl_c_recv()` 给自己留一个方便的“停止开关”呢？**
**建议：在所有需要交互式停止的 Funboost 脚本末尾，都加上这一行。**
## 6.25b.5 顺便说一下，文档中其他地方提到的 `ctrl_c_recv` 和 apscheduler 的定时器之间的关系
很多开发者（包括AI）会疑惑：文档其他地方说用定时任务时必须加 `ctrl_c_recv()` 或 `while 1: time.sleep(10)`，否则会报错 `RuntimeError: cannot schedule new futures after interpreter shutdown`，但这里又说可以不加，到底哪个是对的？
其实答案很简单：**取决于你用的 Funboost 版本**。
### 6.25b.5.1 APScheduler 的两种原生定时器
众所周知，APScheduler 有两种常用定时器：
- **`BlockingScheduler`**：阻塞型。调用 `start()` 后，代码会卡住不动，后续代码根本没有机会运行。
- **`BackgroundScheduler`**：非阻塞型。调用 `start()` 后，代码会继续往下执行，定时任务在**守护线程**中运行。
### 6.25b.5.2 早期版本（2025年之前）的问题
Funboost 早期为了能让用户连续写启动多个函数消费 `fun_xx.consume()` 和连续启动多个定时器 `scheduler_xx.start()` 而不阻塞，选择了继承 `BackgroundScheduler` 作为定时器基类。
原生的 `BackgroundScheduler` 有一个特点：它的工作线程是**守护线程**（`_daemon=True`）。
当你的脚本执行完所有代码，主线程结束后，守护线程会被 Python 强行终止。此时如果定时器正在尝试向线程池提交新任务，就会触发：
```python
RuntimeError: cannot schedule new futures after interpreter shutdown
```
所以早期文档里才会反复强调：**如果使用了定时任务，必须在脚本末尾加上 `ctrl_c_recv()` 或 `while 1: time.sleep(10)`，让主线程一直活着，避免报错。**
### 6.25b.5.3 2025年之后的改进
现在 `FunboostBackgroundScheduler` 已经重写了 `_daemon` 属性，**显式将其设为 `False`**。这意味着：
- 定时器工作线程变成了**非守护线程**。
- 即使主线程执行完毕，这些非守护线程也会让进程继续保持运行，不会触发上述报错。
**所以从 2025 年以后的 Funboost 版本开始，即使你用了定时任务，也完全不需要加 `ctrl_c_recv()` 或 `while 1: time.sleep(10)`，程序就能正常运行且不会报错。**
### 6.25b.5.4 三种定时器的行为对比
| 定时器类型 | 是否阻塞 | 线程类型 | 不加 `ctrl_c_recv` 会怎样？ |
| :--- | :--- | :--- | :--- |
| **原生 `BlockingScheduler`** | 阻塞 | - | `start()` 后卡住，后续代码根本跑不到，没有这个问题 |
| **原生 `BackgroundScheduler`** | 非阻塞 | **守护线程** | 主线程结束后，守护线程被强杀，会报 `RuntimeError` |
| **`FunboostBackgroundScheduler`** | 非阻塞 | **非守护线程** | 主线程结束后，非守护线程继续运行，**不会报错** |
### 6.25b.5.5 总结
- 如果你用的是 **2025年之后的 Funboost**，使用定时任务时**可以不加** `ctrl_c_recv()`，程序会一直运行，定时器也能正常工作。
- 如果你用的是**旧版本**，或者不确定版本，**建议还是加上** `ctrl_c_recv()` 或 `while 1: time.sleep(10)` 保平安。
- 当然，即使在新版本中，加上 `ctrl_c_recv()` 也**没有副作用**，还能让你方便地用 `Ctrl + C` 停止程序，所以加上也没什么坏处。
## 6.26 ASYNC 并发模式,异步函数操作数据库/aiohttp连接池报错 `attached to a different loop`(是用户不看文档和boost的入参说明造成)
- **第一性原理**：对于跨线程玩loop 协程，如果用户经常这样编程的一定遇到过这种经典的坑，遇到了才能使自己的asyncio loop知识更上一层楼；
如果一个pythoner，一直只在主线程玩一个loop，是很难真正掌握asyncio的知识的，尤其是不懂loop。
- 使用指定的 `specify_async_loop` 即可解决.
这个放在自问自答 6.26 章节有源码注释演示,太多人不看文档,不看boost装饰器的入参解释说明.不看 `specify_async_loop` 入参解释造成的.
- asyncio的经典报错 `attached to a different loop` 和 `context manager should be used inside a task` 和`attached to a different loop` 报错原因:
这个错误通常发生在 对象（如 Future、Task、协程）在一个事件循环中创建，却在另一个事件循环中被使用。
- `RuntimeError: Timeout context manager should be used inside a task` 报错原因:
异步上下文管理器（如timeout）不在Task中运行
- 很多人在async def消费函数中去操作http连接池 发请求,操作数据库连接池查询数据,在funboost中报错.
根本原因是用户不传递指定 `specify_async_loop`, 如果不传递,funboost是独立线程中启动了一个新的loop,
用户的连接池绑定的是主线程的loop,如果使用子线程的新loop去使用这个连接池查询数据库,那就会导致报错.
**只在主线程中操作`asyncio`协程的pythoner,永远无法理解这个知识点,需要用户多练习在子线程去操作数据库连接池或者http连接池,才能踩坑积累经验.**
用户可以结合看4b.3章节
**用户需要始终知道 ASYNC 并发模式 第一性原理,才知道怎么根本解决问题**
切记切记,funboost 的 ASYNC 并发模式的核心底层原理,
funboost,当用户使用 ASYNC 并发模式时候,是自动使用 AsyncPoolExecutor 来执行用户函数,
AsyncPoolExecutor 源码在 funboost/concurrent_pool/async_pool_executor.py
AsyncPoolExecutor 原理是启动了一个线程,这个线程会使用传递的specify_async_loop,如果不传递就会新创建1个loop,
这个loop会运行指定的 concurrent_num 个cron 协程,去运行用户的消费函数逻辑
也就是说AsyncPoolExecutor线程只有一个,loop只有一个,真正的一个loop并发运行cron协程
用 loop2 去运行 loop1 创建的task,不报错才怪
- **即使是funboost 的 thereding 并发模式也可以直接运行async def 函数,funboost对async的支持完爆celery**
funboost 的 thereding 并发模式也可以直接@boost装饰加到async def 函数,
因为funboost是特制的神级别线程池,能自动运行async def 函数.
但这个模式下简单粗暴,会开启 concurrent_num 个线程,每个线程有自己的loop,去运行用户的async函数,
相当于是每个loop只并发运行一个cron协程,不是真asyncio级别并发
相当于是有无数线程,每个线程一个loop,去运行用户的async函数
funboost的 thereding 并发模式运行async,就是celery 任务中强行使用 asyncio.new_event_loop().run_until_complete(async函数)
但是funboost是特制的神级别线程池 FlexibleThreadPool,这个线程池自动可以运行async def 函数,
无需用户为了运行async def 函数,去手写一个脱了裤子放屁的 同步def 函数,
也就是用户无需这种脱了裤子放屁的写法:
@boost(BoosterParams(queue_name='test_async_queue2', concurrent_mode=ConcurrentModeEnum.THREADING))
def 同步fun(x,y)
asyncio.new_event_loop().run_until_complete(async异步函数(x,y)) #同步里面调用异步函数
用户可以直接使用THREADING 并发模式加到 async def async异步函数:
@boost(BoosterParams(queue_name='test_async_queue2', concurrent_mode=ConcurrentModeEnum.THREADING))
async def async异步函数(x,y)
pass
celery的运行协程才需要脱落裤子放屁再加个同步函数里面run_until_complete调用异步函数,才能@app.task
### 6.26.1 演示aiohttp连接池,解决报错 `RuntimeError: context manager should be used inside a task`
演示aiohttp连接池,解决报错代码如下,用户需要看里面的每个代码注释.
```python
"""
此脚本主要是演示，在funboost自动的单独的子线程的loop中,如何使用asyncio的异步socket池，来发送异步请求。
核心是要把主线程的loop传递到子线程中,让子线程和主线程使用同一个loop来运行异步函数,
在funboost 中 通过 specify_async_loop=主线程loop 来传递.
跨线程的不同loop,使用在主线程的loop生成的异步连接池 来发请求是不行的.
不管是任何三方包的socket池,例如 aiomysql aioredis aiohttp httpx 创建的http或者数据库连接池,
都不能在子线程的loop的异步函数中直接去使用这个连接池中的连接发请求.
有些异步三方包的连接池能直接在子线程的loop去使用连接发请求或查询数据库而不报错,是因为惰性生成的.
"""
"""
用户一定要搞清楚 线程和loop的绑定关系
一定要知道为什么不同的loop,不能操作同一个异步连接池发请求或者查询数据库.
用户一定要多写子线程的loop调用连接池发请求测试demo,这个和funboost本身无关,
用户的asyncio知识体系太差,用户只会在主线程使用loop,导致对loop和线程绑定关系不懂,对不同loop操作一个连接池不懂.
在子线程运行异步函数的loop,比在主线程的loop运行难得多,坑也更多.用户需要多写demo例子测试练习,多问ai大模型.
"""
# 例如子线程的loop去使用主线程loop绑定的http连接池发请求,会报错如下. 数据库连接池同理也会报错.
"""
Traceback (most recent call last):
File "D:\codes\funboost\funboost\consumers\base_consumer.py", line 929, in _async_run_consuming_function_with_confirm_and_retry
rs = await corotinue_obj
File "D:\codes\funboost\test_frame\test_async_consumer\test_child_thread_loop_use_asyncio_sokcet_pool.py", line 64, in async_f2
async with ss.request('get', url=url) as resp: # 如果是这样请求，boost装饰器必须指定specify_async_loop，
File "D:\ProgramData\Miniconda3\envs\py39b\lib\site-packages\aiohttp\client.py", line 1425, in __aenter__
self._resp: _RetType = await self._coro
File "D:\ProgramData\Miniconda3\envs\py39b\lib\site-packages\aiohttp\client.py", line 607, in _request
with timer:
File "D:\ProgramData\Miniconda3\envs\py39b\lib\site-packages\aiohttp\helpers.py", line 636, in __enter__
raise RuntimeError("Timeout context manager should be used inside a task")
RuntimeError: Timeout context manager should be used inside a task
"""
from funboost import boost, BrokerEnum,ConcurrentModeEnum,ctrl_c_recv
import asyncio
import aiohttp
import time
url = 'http://mini.eastday.com/assets/v1/js/search_word.js'
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop) # 这是重点，设置主线程默认的loop 为你创建的loop
# 这是重点,ss和主线程的loop绑定了. # 新版本aiohttp必须放在协程函数中去实例化，不能全局变量实例化，参考下面的 aiomysql
ss = aiohttp.ClientSession(loop=loop)
@boost(BoosterParams(queue_name='test_async_queue1', concurrent_mode=ConcurrentModeEnum.ASYNC, broker_kind=BrokerEnum.REDIS,
log_level=10,concurrent_num=3,
specify_async_loop=loop, # specify_async_loop传参是核心灵魂代码,不传这个还要使用主loop绑定的连接池就会报错.
is_auto_start_specify_async_loop_in_child_thread=False,
))
async def async_f1(x):
"""
这个函数是自动被funboost在一个单独的子线程中的loop运行的,loop会并发运行很多协程来执行async_f1的逻辑
用户最最需要明白的是,在用funboost的 ConcurrentModeEnum.ASYNC时候,你不是在主线程中操作的异步函数,而是子线程的loop中调用的.
假设如果是在主线程中去运行的,你怎么可能连续丝滑启动多个函数消费 f1.consume() f2.consume() f3.consume() ? 用脑子想想就知道不是主线程去调用异步函数的.
"""
# 如果是async with ss.request('get', url=url)使用主线程loop的连接池发请求，boost装饰器必须指定 specify_async_loop，
# 如果你不使用ss连接池,而是 async with aiohttp.request('GET', 'https://httpbin.org/get') as resp: 那就不需要指定 specify_async_loop
async with ss.request('get', url=url) as resp:
text = await resp.text()
print('async_f1', x, resp.url, text[:10])
await asyncio.sleep(5)
return x
@boost(BoosterParams(queue_name='test_async_queue2', concurrent_mode=ConcurrentModeEnum.ASYNC, broker_kind=BrokerEnum.REDIS,
log_level=10,concurrent_num=3,
# specify_async_loop=loop, # specify_async_loop传参是核心灵魂代码,连接池不传这个还要使用主loop绑定的连接池就会报错.
is_auto_start_specify_async_loop_in_child_thread=False,
))
async def async_f2(x):
# 如果是async with ss.request('get', url=url)使用主线程loop的连接池发请求，boost装饰器必须指定 specify_async_loop，
# 如果你不使用ss连接池,而是 async with aiohttp.request('GET', 'https://httpbin.org/get') as resp: 那就不需要指定 specify_async_loop
async with aiohttp.request('get', url=url) as resp:
text = await resp.text()
print('async_f2', x, resp.url, text[:10])
await asyncio.sleep(5)
return x
async def do_req(i):
async with ss.request('get', url=url) as resp: # 如果是这样请求，boost装饰器必须指定specify_async_loop，
text = await resp.text()
print(f'主线程的loop运行的{i}:',text[:10])
await asyncio.sleep(3)
if __name__ == '__main__':
async_f1.clear()
async_f2.clear()
for i in range(10):
async_f1.push(i)
async_f2.push(i*10)
async_f1.consume()
async_f2.consume()
# time.sleep(5) 加这个是测试主线程的loop和子线程loop谁先启动,造成的影响,如果子线程的specify_async_loop先启动,主线程下面的 loop.run_forever() 会报错已启动 RuntimeError: This event loop is already running
main_tasks = [loop.create_task(do_req(i)) for i in range(20)]
loop.run_forever() # 如果你除了要在funboost运行异步函数,也要在自己脚本调用,那么装饰器配置 is_auto_start_specify_async_loop_in_child_thread=False,,自己手动 启动loop.run_forever()
ctrl_c_recv()
```
如果消费async def的消费函数中:
如果是async with ss.request('get', url=url)使用主线程loop的连接池发请求，boost装饰器必须指定 specify_async_loop，
如果你不使用ss连接池,而是 async with aiohttp.request('GET', 'https://httpbin.org/get') as resp: 那就不需要指定 specify_async_loop
用户一定要好好体会下原因,为什么会这样,才能更懂asyncio
### 6.26.2 演示 aiomysql 连接池在funboost使用,解决 `attached to a different loop`
用户需要认真看代码里面的文字注释,为什么出现 `attached to a different loop` 经典报错
代码例子:
```python
"""
此脚本演示, funboost的子线程的loop中怎么操作aiomysql 连接池.
演示了2种方式使用aiomysql连接池
方式一:
非常推荐方式一
async_aiomysql_f1 使用主线程的连接池,.
核心灵魂代码是 async_aiomysql_f1 的装饰器需要传参specify_async_loop=主线程loop
如果 async_aiomysql_f1 不指定specify_async_loop,就会出现经典报错 attached to a different loop ,
子线程的loop去操作主线程loop的连接池,这是大错特错的.
有的人压根不懂主线程loop和子线程loop,还非要装逼使用asyncio编程生态,
不精通asyncio编程生态的人应该老老实实使用同步多线程编程生态,简单多了.
因为funboost的线程池 FlexibleThreadPool能自动扩缩,
能自动缩容是吊打内置线程池 concurrent.futures.threadpoolexecutor的神级别操作,
FlexibleThreadPool 性能比官方内置线程池提高了250%
-----------------------------------------------------------------------
方式2:
不推荐方式2，浪费连接池宝贵资源。
async_aiomysql_f2_use_thread_local_aio_mysql_pool 使用 thread local 级别的全局变量连接池
这样子线程的loop避免了操作主线程loop的aiomysql连接池, 不会出现张冠李戴 attached to a different loop
"""
import threading
from funboost import boost, BrokerEnum, ConcurrentModeEnum, ctrl_c_recv, BoosterParams
import asyncio
import time
import aiomysql
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop) # 这是重点
DB_CONFIG = {
'host': 'localhost',
'port': 3306,
'user': 'root',
'password': '123456',
'db': 'testdb',
'charset': 'utf8mb4',
'autocommit': True
}
g_aiomysql_pool : aiomysql.Pool
async def create_pool():
pool = await aiomysql.create_pool(**DB_CONFIG, minsize=1, maxsize=10,) # 这些是重点.
global g_aiomysql_pool
g_aiomysql_pool = pool
return pool
# 如果 async_aiomysql_f1 不指定specify_async_loop,就会出现经典报错 attached to a different loop ,子线程的loop去操作主线程loop的连接池
r"""
Traceback (most recent call last):
File "D:\codes\funboost\test_frame\test_async_consumer\test_child_thread_loop_aiomysql_pool.py", line 46, in async_aiomysql_f1
await cur.execute("SELECT now()")
File "D:\ProgramData\Miniconda3\envs\py39b\lib\site-packages\aiomysql\cursors.py", line 239, in execute
await self._query(query)
File "D:\ProgramData\Miniconda3\envs\py39b\lib\site-packages\aiomysql\cursors.py", line 457, in _query
await conn.query(q)
File "D:\ProgramData\Miniconda3\envs\py39b\lib\site-packages\aiomysql\connection.py", line 469, in query
await self._read_query_result(unbuffered=unbuffered)
File "D:\ProgramData\Miniconda3\envs\py39b\lib\site-packages\aiomysql\connection.py", line 683, in _read_query_result
await result.read()
File "D:\ProgramData\Miniconda3\envs\py39b\lib\site-packages\aiomysql\connection.py", line 1164, in read
first_packet = await self.connection._read_packet()
File "D:\ProgramData\Miniconda3\envs\py39b\lib\site-packages\aiomysql\connection.py", line 609, in _read_packet
packet_header = await self._read_bytes(4)
File "D:\ProgramData\Miniconda3\envs\py39b\lib\site-packages\aiomysql\connection.py", line 657, in _read_bytes
data = await self._reader.readexactly(num_bytes)
File "D:\ProgramData\Miniconda3\envs\py39b\lib\asyncio\streams.py", line 723, in readexactly
await self._wait_for_data('readexactly')
File "D:\ProgramData\Miniconda3\envs\py39b\lib\asyncio\streams.py", line 517, in _wait_for_data
await self._waiter
RuntimeError: Task
> got Future attached to a different loop
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "D:\codes\funboost\funboost\consumers\base_consumer.py", line 929, in _async_run_consuming_function_with_confirm_and_retry
rs = await corotinue_obj
File "D:\codes\funboost\test_frame\test_async_consumer\test_child_thread_loop_aiomysql_pool.py", line 48, in async_aiomysql_f1
print(res)
File "D:\ProgramData\Miniconda3\envs\py39b\lib\site-packages\aiomysql\utils.py", line 139, in __aexit__
await self._pool.release(self._conn)
RuntimeError: Task > got Future > attached to a different loop
"""
"""
方式一:
消费函数使用全局g_aiomysql连接池,每个线程都使用主线程的loop来使用连接池查询,所以@boost需要指定specify_async_loop为主线程的loop
"""
@boost(BoosterParams(queue_name='async_aiomysql_f1_queue', concurrent_mode=ConcurrentModeEnum.ASYNC, broker_kind=BrokerEnum.REDIS,
log_level=10,concurrent_num=3,
specify_async_loop=loop, # specify_async_loop传参是核心灵魂代码,不传这个还要使用主loop绑定的连接池就会报错.
is_auto_start_specify_async_loop_in_child_thread=True,
))
async def async_aiomysql_f1(x):
await asyncio.sleep(5)
async with g_aiomysql_pool.acquire() as conn:
async with conn.cursor() as cur:
await cur.execute("SELECT now()")
res = await cur.fetchall()
print(res)
"""
方式二:
消费函数使用thread_local线程级别全局变量,每个消费函数的那个子线程的loop使用的是线程级别自己的连接池,
这样就避免了子线程的loop去用 主线程绑定的aiomysqlpool查询数据库,导致报错
强烈不推荐方式二，这会给每个消费函数创建一个mysql 连接池，
因为如果不显式传递specify_async_loop，默认是每个booster用一个单独的线程启动loop，每个消费函数有自己的loop，浪费mysql连接数量的宝贵资源。
你的booster消费函数越多越不利于mysql连接池复用。
"""
thread_local = threading.local()
async def create_pool_thread_local():
if hasattr(thread_local, 'aiomysql_pool'):
return thread_local.aiomysql_pool
pool = await aiomysql.create_pool(**DB_CONFIG, minsize=1, maxsize=5, ) # 这些是重点.
setattr(thread_local, 'aiomysql_pool', pool)
print('创建了线程级别 threadlocal的 aiomysql连接池')
return pool
@boost(BoosterParams(queue_name='async_aiomysql_f2_queue', concurrent_mode=ConcurrentModeEnum.ASYNC, broker_kind=BrokerEnum.REDIS,
log_level=10,concurrent_num=3,
# specify_async_loop=loop, # specify_async_loop 不需要传递,因为子线程的loop有自己的pool,不使用主线程的pool去查数据库
is_auto_start_specify_async_loop_in_child_thread=True,
))
async def async_aiomysql_f2_use_thread_local_aio_mysql_pool(x):
await asyncio.sleep(5)
aiomysql_pool_thread_local = await create_pool_thread_local()
async with aiomysql_pool_thread_local.acquire() as conn:
async with conn.cursor() as cur:
await cur.execute("SELECT now()")
res = await cur.fetchall()
print(res)
if __name__ == '__main__':
loop.run_until_complete(create_pool()) # 先创建pool,aiomysql的pool不能直接在模块级全局变量直接生成.
async_aiomysql_f1.clear()
async_aiomysql_f2_use_thread_local_aio_mysql_pool.clear()
for i in range(10):
async_aiomysql_f1.push(i)
async_aiomysql_f2_use_thread_local_aio_mysql_pool.push(i*10)
async_aiomysql_f1.consume()
async_aiomysql_f2_use_thread_local_aio_mysql_pool.consume()
ctrl_c_recv()
```
### 6.26.3 演示子线程怎么正确的使用aiomysql连接池的本质(和funboost无关但原理相通)
有些人报错 `RuntimeError: attached to a different loop`,以为是 funboost的bug,其实是因为压根不懂asyncio的loop知识.
下面这个代码没有使用funboost来消费async 函数,和funboost原理一样,子线程操作主线程的异步连接池
```python
"""
演示子线程中的loop怎么正确操作 异步aio连接池
正解就是子线程要使用 asyncio.run_coroutine_threadsafe(aio_do_select(1), 主线程的loop)
错误做法就是,子线程生成自己的loop,却用主线程的aio连接池查询数据库,造成经典错误 RuntimeError: attached to a different loop
funboost的 AsyncPoolExecutor 就是在子线程运行的loop ,需要懂这个例子,才知道为什么操作aio连接池时候, funboost装饰器为什么要把主loop传给 specify_async_loop
"""
import threading
import asyncio
import time
import aiomysql
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop) # 这是重点
DB_CONFIG = {
'host': 'localhost',
'port': 3306,
'user': 'root',
'password': '123456',
'db': 'testdb',
'charset': 'utf8mb4',
'autocommit': True
}
g_aiomysql_pool : aiomysql.Pool
async def create_pool():
print(f'主线程 threading id: {threading.get_ident()}')
pool = await aiomysql.create_pool(**DB_CONFIG, minsize=1, maxsize=10,) # 这些是重点.
global g_aiomysql_pool
g_aiomysql_pool = pool
return pool
async def aio_do_select(i):
async with g_aiomysql_pool.acquire() as conn:
async with conn.cursor() as cur:
await cur.execute("SELECT now()")
res = await cur.fetchall()
print(res)
def run_do_select_in_child_thread(loopx:asyncio.AbstractEventLoop):
""" 查询数据库正确的做法,
使用主线程的loop去查数据库.
"""
print(f'子线程 threading id: {threading.get_ident()}')
asyncio.run_coroutine_threadsafe(aio_do_select(1), loopx) # 这是正解
def run_do_select_in_child_thread_error():
"""查询数据库错误的写法, 用子线程自己的loop去调用 主线程loop绑定的g_aiomysql_pool,
造成经典错误 RuntimeError: attached to a different loop
"""
print(f'子线程 threading id: {threading.get_ident()}')
loop = asyncio.new_event_loop()
loop.run_until_complete(aio_do_select(1))
if __name__ == '__main__':
loop.run_until_complete(create_pool())
threading.Thread(target=run_do_select_in_child_thread, args=(loop,)).start()
threading.Thread(target=run_do_select_in_child_thread_error, ).start()
loop.run_forever()
```
## 6.28 日志提示 是掉线或关闭消费者的 和 重新放入掉线消费者未消费确认的任务 ,是正常的提示.
例如提示下面,是完全正常的,是确认消费功能的 重回孤儿消息.
```
2025-08-18 12:37:19-(10.0.133.57,LAPTOP-7V78BBO2)-[p58428_t59904] - funboost.RedisConsumerAckAble--test_cost_long_time_fun_queue888 - "confirm_mixin.py:83" - _requeue_tasks_which_unconfirmed - WARNING - no_task_id - test_cost_long_time_fun_queue888__unack_id_LAPTOP-7V78BBO2_10.0.133.57_2025-08-18 12-36-02_46752_2024011500368 是掉线或关闭消费者的
2025-08-18 12:37:19-(10.0.133.57,LAPTOP-7V78BBO2)-[p58428_t59904] - funboost.RedisConsumerAckAble--test_cost_long_time_fun_queue888 - "confirm_mixin.py:87" - _requeue_tasks_which_unconfirmed - WARNING - no_task_id - 从 test_cost_long_time_fun_queue888__unack_id_LAPTOP-7V78BBO2_10.0.133.57_2025-08-18 12-36-02_46752_2024011500368 向 test_cost_long_time_fun_queue888 重新放入掉线消费者未消费确认的任务 {"x":666,"extra":{"task_id":"test_cost_long_time_fun_queue888_result:7ff6270e-5669-4ee4-8c48-d5b9ba39e3d9","publish_time":1755491762.9382,"publish_time_format":"2025-08-18 12:36:02"}}
```
```
在在选择redis可确认消息的(非BrokerEnum.REDIS)中间件时候,
例如你应该几把脚本反复部署启动2次kill了其中一个,或者即使只有一个脚本,但是重启后,就会提示这样啊,这是提示把之前kill的消费者进程所持有的待确认消费的消息,重回工作队列, 确保只有确认消费的消息才会真的不再次被调度运行.
那些运行到一半的函数消息和内存中缓存的一部分消息,在kill后,都会被重回消息队列.
```
```
用户如果不需要确认消费功能,就使用 BrokerEnum.REDIS 模式,这个是no ack 模式.
以下redis并发模式都是need ack 确认消费的,消息从队列取出来后,只要消息没确认消费,进程被关闭后,孤儿消息下次还会被重回队列
REDIS_ACK_ABLE
REIDS_ACK_USING_TIMEOUT
REDIS_PRIORITY
REDIS_STREAM
RedisBrpopLpush
这些在 BrokerEnum 源码注释中 以及 教程文档中反复说了无数次,用户太懒了不看
```
## 6.29 funboost 推荐用户使用什么并发模式？
在 Funboost 中，并发模式的选择并非一道选择题，而是一套层层递进的“性能方程式”。框架提供了多种模式，但在 99% 的情况下，你只需要遵循以下推荐原则即可。
### 6.29.1 默认推荐：当前进程内 + 多线程模式 (最省心、最稳定、最好用)
**适用场景**：绝大多数的常规后端任务，如 API 请求、爬虫、写数据库、发送邮件等 IO 密集型操作。
**使用方式**：什么都不用改。`@boost` 的 `concurrent_mode` 默认就是 `ConcurrentModeEnum.THREADING`，直接 `task_fun.consume()` 启动即可。
**为什么这是最推荐的模式？**
因为在这个模式下，支撑并发的底层**不是** Python 官方自带的 `concurrent.futures.ThreadPoolExecutor`，而是 Funboost 作者纯手工打造的神级线程池：**`FlexibleThreadPool`（智能弹性线程池）**。
它吊打官方原生线程池的核心优势在于：
1. **自动智能缩容**：官方线程池是“只进不出”的貔貅，无论任务多么稀疏，官方线程池会盲目增加到最大线程数；并且后来长期没啥任务了，线程也一直挂在内存里。而 `FlexibleThreadPool` 引入了 `KeepAliveTime` 机制，在任务稀疏时会自动销毁多余线程，释放系统资源。
2. **极度克制的扩张**：哪怕你配置了 `concurrent_num=500`，如果你的任务很稀疏，它就绝对不会去新开不必要的线程。
3. **兼容 `async def` 函数**：你可以直接把 `@boost(concurrent_mode=THREADING)` 加在异步函数上！框架会自动在线程池的每个线程里启动临时的 event loop 来运行协程，免去了你手动改写同步包装函数的痛苦（这一点完爆 Celery）。
**一句话总结**：这是兼顾了“极低心智负担”与“极高 IO 吞吐”的王牌模式。
### 6.29.2 高性能推荐：多进程叠加并发模式 (性能炸裂)
**适用场景**：
1. **CPU 密集型任务**（如视频转码、图像处理、复杂数学计算），需要突破 Python 的 GIL 锁。
2. **超高并发的 IO 任务**，单进程的线程数开到极限依然无法满足吞吐需求。
**使用方式**：
**不要**去修改 `@boost` 装饰器里的 `concurrent_mode`（依然保持默认的多线程）。而是在启动消费时，使用 **`task_fun.multi_process_consume(n)`** （可简写为 `task_fun.mp_consume(n)`）。
**核心原理**：
这是 Funboost 的特色“叠加态”并发。例如执行 `task_fun.mp_consume(4)`：
* 框架会先利用操作系统的多进程（Process）开出 4 个独立的 Worker 进程（完美利用 4 核 CPU）。
* 在每个进程内部，依然使用 `FlexibleThreadPool` 开启比如 100 个多线程。
* 总并发能力 = 4 进程 × 100 线程 = 400 个并发单元。
**为什么不直接用纯多进程模式？**
如果不开线程，纯靠开 400 个进程去应对并发，操作系统的内存会瞬间被撑爆，而且进程切换的开销极大。Funboost 的“少量进程 + 海量线程”模式，是用最低廉的内存代价，榨干机器性能的最佳实践。
#### 6.29.2.2 ⚠️ 极度重要：多进程在 Linux 上建议显式设置 spawn 模式，抛弃默认的 fork 模式
**多进程下，spawn 模式相比 fork 模式，在“环境一致性”和“避免内存污染”上有巨大优势。**
如果你使用了多进程叠加并发，强烈建议在你的启动脚本**最顶端**加上以下代码：
```python
import multiprocessing
multiprocessing.set_start_method('spawn', force=True)
```
**为什么 Funboost 强烈推崇 `spawn` 而嫌弃 `fork`？**
| 优势 | 核心要点 |
| :--- | :--- |
| **环境一致性** | Windows 只有 `spawn`，Linux 默认 `fork`，显式设置 `spawn` 可避免"在我电脑上明明跑得好好的"困境 |
| **架构先进性** | Funboost "少量进程 + 海量线程"理念，不需要为了启动快而牺牲安全性 |
| **内存纯粹性** | `spawn` 干净独立，消灭 `MySQL Server has gone away` 等僵尸连接问题 |
| **开发体验** | 消灭 `@worker_process_init` 等反人类钩子，无需手动重新初始化连接 |
| **调试友好** | `spawn` 每次启动都是干净的 Python 进程，错误堆栈清晰可溯，不会被父进程"遗传病"干扰 |
**详细说明：**
1. **抹平开发与生产的环境差异**
Windows 只有 `spawn` 模式，而 Linux 默认是 `fork`。如果你在 Linux 不显式设置 `spawn`，你的代码在 Windows 开发机和 Linux 生产机上的行为将不一致，极易让新手陷入"在我电脑上明明跑得好好的"的懵逼困境。
2. **拒绝 Celery 的历史包袱**
Celery 默认是纯多进程模式，且无法叠加细粒度并发（不能进程套线程），所以它为了启动快而依赖 `fork`。但 Funboost 理念是"少量进程 + 海量线程"，进程启动的极小开销完全可以忽略，根本不需要为了图快去用 `fork`。
3. **保持内存纯粹，消灭僵尸连接**
`fork` 是"克隆"父进程的内存（包括父进程已经建立的数据库连接池、线程锁等）。这会导致子进程在查询数据库时发生经典的 `MySQL Server has gone away` 报错或死锁。
> 💡 **一句话理解**：用 `spawn` 启动 4 个进程，就等于你在控制台手动敲了 4 次 `python xx.py`。每个进程重新执行一遍初始化代码，互不干扰。
4. **消灭反人类的钩子函数**
正因为 Celery 饱受 `fork` 内存污染的折磨，才被迫发明了 `@worker_process_init` 这种极其操蛋的钩子，强迫小白去学习"如何在这个钩子里重新初始化数据库连接"。**在 Funboost + `spawn` 的世界里，这种恶心的问题从物理层面上就被消灭了。**
5. **更好的错误追踪和调试体验**
`fork` 的子进程会继承父进程的所有内存状态，包括异常状态、信号处理器等。当子进程出现问题时，你很难定位是父进程继承来的"遗传病"还是子进程自己产生的"新病"。
而 `spawn` 是全新的 Python 解释器进程，**每次启动都是干净的 slate**，错误堆栈清晰可溯，调试时不会被父进程的"历史包袱"干扰。
### 6.29.3 极客推荐：纯 Asyncio 协程模式 (高手的玩具)
**适用场景**：
你是一个重度的 `asyncio` 玩家，你的项目已经全面拥抱了异步生态（如使用 FastAPI），并且你在消费函数中必须使用 `aiohttp`、`aiomysql`、`aioredis` 等异步连接池。
**使用方式**：
必须在 `@boost` 中显式指定：
```python
@boost(BoosterParams(
queue_name='my_queue',
concurrent_mode=ConcurrentModeEnum.ASYNC,
specify_async_loop=my_main_loop # 强烈建议阅读 6.26 章节，懂这个参数的才是真懂 asyncio
))
async def my_task(x):
pass
```
**为什么普通人不推荐用纯 ASYNC 模式？**
因为 asyncio 在 Python 中是一把“双刃剑”。
* 一旦你选择了 ASYNC 模式，你的函数里就**绝对不能**出现任何阻塞的同步代码（比如普通的 `requests.get` 或 `time.sleep`），否则整个 Event Loop 都会被卡死，并发瞬间变成串行，引发灭顶之灾。
* 更要命的是“跨线程使用 Loop”的经典大坑（即抛出 `attached to a different loop` 错误）。如果你不懂得如何把主线程的数据库连接池通过 `specify_async_loop` 传递给 Funboost 子线程的 Loop（详见文档 6.26 章节），你会死得很惨。
**总结**：如果你没经过系统的 asyncio 毒打，老老实实用 6.29.1 的多线程模式就好。
### 6.29.4 特殊场景：单线程串行模式 (Single Thread)
**适用场景**：
1. **极度排斥并发冲突的任务**：你的任务绝对不能并发执行，比如按严格顺序更新某一行数据库记录，或者操作某个不支持并发的本地文件/老旧系统。
2. **要求“绝对公平”的跨机器负载均衡**：你的服务绝对不能容忍“预先从 Broker 抢一堆消息缓冲在内存里”，必须保证多台机器公平、按需地抢夺消费机会。
**使用方式**：
```python
@boost(BoosterParams(
queue_name='my_queue',
concurrent_mode=ConcurrentModeEnum.SINGLE_THREAD
))
def my_task(x):
pass
```
**⚠️ 深度辨析：`SINGLE_THREAD` 和 `THREADING + concurrent_num=1` 是一回事吗？**
**绝对不是！这是两个完全不同的物种。**
* **`THREADING` 模式（即使你设置了 `concurrent_num=1`）**：
底层依然会启动 `FlexibleThreadPool`。只要有线程池，就会存在一个内存缓冲队列（`_work_queue`）。这意味着主消费线程会**贪婪地从 Broker（如 Redis）中疯狂拉取几十上百条消息，囤积到自己进程的内存里**，然后再由那 1 个工作线程慢慢消化。
**后果**：如果你部署了 A、B 两台机器，A 机器可能瞬间把队列里的 50 条消息全抢到了自己的内存里，导致 B 机器在一旁无所事事（闲死），而 A 机器却要花很久才处理完（忙死），彻底破坏了多机的负载均衡。
* **`SINGLE_THREAD` 模式**：
框架**没有**线程池缓冲队列！它会极其克制、死板地执行：去 Broker 拉取 **1 条**消息 -> 执行完毕 ->再去 Broker 拉取下 **1 条**消息。
**后果**：无论你部署多少台机器，大家全凭手速，做完一个拿一个，绝不超售囤货，实现了**绝对公平的分布式多机负载均衡**。稳如老狗。
### 6.29.5 至于 gevent 和 eventlet 并发模式，直接忽略就好
虽然 funboost 并发模式支持了 `gevent` 和 `eventlet`，但**不推荐**使用，你没有必要研究它们。
## 6.30 funboost 如何实配置触发失败告警和恢复告警
funboost 提供了多种灵活的失败告警方式，满足从业务级即时通知到运维级聚合监控的不同需求。以下介绍5种常用方案。
### 6.30.1 使用内置告警 Mixin（AlertNotifier）
**适用场景**：当消费函数连续失败次数或错误率超过阈值时，实时发送钉钉、企业微信、飞书或自定义 Webhook 告警。
- **实现原理**：通过 `AlertNotifierConsumerMixin` 拦截任务执行结果，基于“连续失败计数”或“滑动窗口错误率”策略触发告警，并提供去重机制和恢复通知。
- **配置方式**：推荐使用预配置的 `AlertNotifierBoosterParams`，只需在 `user_options` 中填写 `alert_options` 参数。
- **告警渠道支持 企业微信 钉钉 飞书，只需要提供webhook**
详细参数说明见 `funboost/contrib/override_publisher_consumer_cls/alert_notifier_mixin.py` 。
#### 6.30.1 AlertNotifierConsumerMixin 发送警告到 微信 钉钉 飞书
```python
from funboost import boost, BrokerEnum
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, # 连续失败5次触发
'alert_app': 'wechat', # 企业微信机器人
'webhook_url': 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx',
'alert_interval': 300, # 5分钟内不重复告警
}
}
))
def my_task(x):
# 业务逻辑
...
```
#### 6.30.1 AlertNotifierConsumerMixin 发送警告到任意渠道
```
用户自定义告警发送钩子，当 alert_app='custom' 时自动调用。
继承 AlertNotifierConsumerMixin 并重写此方法，可实现任意告警方式（邮件、短信、企业内部系统等）。
```
```python
class EmailAlertConsumer(AlertNotifierConsumerMixin):
def custom_send_notification(self, message: str):
send_email(to='ops@example.com', subject='任务告警', body=message)
@boost(BoosterParams(
queue_name='my_task',
consumer_override_cls=EmailAlertConsumer,
user_options={'alert_options': {'alert_app': 'custom', 'failure_threshold': 5}},
))
def my_task(x):
...
```
### 6.30.2 通过熔断器钩子自定义告警
**适用场景**：当熔断器状态变化（例如由 `CLOSED` 变为 `OPEN` 或从 `OPEN` 恢复）时，执行自定义告警逻辑。
- **实现原理**：继承 `CircuitBreakerConsumerMixin`，重写 `_on_circuit_open`（熔断触发）和 `_on_circuit_close`（熔断恢复）钩子方法，在其中调用你想要的告警接口。
- **配置方式**：创建自定义 Mixin 类，并在 `consumer_override_cls` 中指定。
```python
from funboost import boost, BoosterParams
from funboost.contrib.override_publisher_consumer_cls.circuit_breaker_mixin import CircuitBreakerConsumerMixin
class MyCircuitBreakerWithAlert(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',
consumer_override_cls=MyCircuitBreakerWithAlert,
user_options={
'circuit_breaker_options': {
'failure_threshold': 5,
'recovery_timeout': 60,
}
}
))
def my_task(x):
...
```
熔断器配置详解见文档 **4b.14** 节及 `circuit_breaker_mixin.py`。
### 6.30.3 基于 Prometheus 指标 + Grafana 告警
**适用场景**：需要聚合监控所有任务的失败率、队列积压等指标，并通过 Grafana 设置告警规则，适合运维监控大屏。
- **实现原理**：funboost 通过 `PrometheusBoosterParams`（单进程）或 `PrometheusPushGatewayBoosterParams`（多进程）将任务指标（`funboost_task_total`、`funboost_queue_msg_count` 等）暴露给 Prometheus，然后在 Grafana 中配置告警。
- **配置方式**：在消费函数上使用 Prometheus 参数类，并启动指标采集服务。
```python
from funboost import boost
from funboost.contrib.override_publisher_consumer_cls.funboost_promethus_mixin import (
PrometheusPushGatewayBoosterParams,
)
@boost(PrometheusPushGatewayBoosterParams(
queue_name='my_task',
user_options={
'prometheus_pushgateway_url': 'localhost:9091',
'prometheus_job_name': 'my_app',
'prometheus_push_interval': 10.0,
}
))
def my_task(x):
...
```
启动后，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`
详细指标说明见文档 **4b.9** 节及 `funboost_promethus_mixin.py`。
### 6.30.4 使用funboost的 MongoAlertMonitor 实现错误告警
**基于监控一段时间内，已保存到mongo的函数执行状态的 success 字段为False的比例或者数量来实现的监控**
这个`MongoAlertMonitor`是分布式汇总监控，所以对于多进程多机器分布式消费错误监控更好。
而`AlertNotifierConsumerMixin`是基于内存计数的监控。
**MongoAlertMonitor监控错误告警的前提条件是必须设置`is_save_status=True`**
```python
@boost(BoosterParams(
...
function_result_status_persistance_conf=FunctionResultStatusPersistanceConfig(
is_save_status=True, # 必须保存函数执行状态到mongo
),
))
```
#### 6.30.4.1 MongoAlertMonitor 使用例子，发到内置实现的 企业微信 钉钉 飞书
**告警渠道支持 企业微信 钉钉 飞书，只需要提供webhook**
```python
# -*- coding: utf-8 -*-
"""
MongoAlertMonitor 告警监控示例
运行前请先启动 t_tasks.py 让任务持续产生成功/失败记录到 MongoDB。
本脚本演示三种使用场景：
场景1：只监控高失败率队列，按报错次数告警
场景2：只监控低失败率队列，按失败率告警
场景3：同时监控两个队列，两种策略叠加（任一满足即告警）
运行时将在 window_seconds=60 秒窗口内统计执行情况，
每 poll_interval=10 秒检查一次，alert_interval=60 秒内不重复发告警。
webhook_url 替换为你自己的企业微信/钉钉/飞书 Webhook 地址。
"""
from funboost.core.mongo_alert_monitor import MongoAlertMonitor
# 导入需要监控的 boosters ，这样才能根据配置，知道查询去监控哪些mongo表
# task_high_failure, task_low_failure 是被 @boost装饰的boosters
from test_frame.test_mongo_alert.t_tasks import task_high_failure, task_low_failure
WECHAT_WEBHOOK = 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxxxx'
# ── 场景1：60 秒内报错次数 >= 5 次 就告警（高失败率队列）────────────────────
MongoAlertMonitor(
boosters=task_high_failure,
alert_app='wechat',
webhook_url=WECHAT_WEBHOOK,
window_seconds=60,
failure_count=5, # 60 秒内报错 >= 5 次触发
poll_interval=10, # 每 10 秒检查一次
alert_interval=60, # 60 秒内不重复发告警（方便测试，生产建议 300+）
).start()
# ── 场景2：60 秒内失败率 >= 50% 且总调用 >= 10 次 就告警（低失败率队列）──────
MongoAlertMonitor(
boosters=task_low_failure,
alert_app='wechat',
webhook_url=WECHAT_WEBHOOK,
window_seconds=60,
errors_rate=0.5, # 失败率 >= 50% 触发
min_calls=10, # 至少 10 次调用才评估
poll_interval=10,
alert_interval=60,
).start()
# ── 场景3：同时监控两个队列，两种策略叠加（任一满足即告警）──────────────────
# 注意：一个 MongoAlertMonitor 实例同时监控多个队列，每个队列告警状态独立
MongoAlertMonitor(
boosters=[task_high_failure, task_low_failure],
# 如果你要监控所有boosters，可以写 boosters=BoostersManager.get_all_boosters()
alert_app='wechat',
webhook_url=WECHAT_WEBHOOK,
window_seconds=60,
failure_count=3, # 任一队列 60 秒内报错 >= 3 次
errors_rate=0.3, # 或 失败率 >= 30%
min_calls=5,
poll_interval=10,
alert_interval=60,
).start()
```
#### 6.30.4.2 MongoAlertMonitor 使用例子，发到自定义的渠道
如果你不是发到企业微信 钉钉 飞书，你如果想发短信 打电话 发邮件 完全可以。
主要是你要写个类，继承 `MongoAlertMonitor` ，重写 `custom_send_notification`方法，类实例化时候要指`alert_app='custom'`
```python
# 示例4：自定义告警通道（发邮件等），继承重写 custom_send_notification
class EmailAlertMonitor(MongoAlertMonitor):
def custom_send_notification(self, message: str):
send_email(to='ops@example.com', subject='任务告警', body=message)
EmailAlertMonitor(
boosters=my_task,
alert_app='custom',
window_seconds=60,
failure_count=10,
).start()
```
### 6.30.5 在funweb 页面中配置告警
funweb页面中也可以配置告警，原理是复用已有的上报到redis中的数据，用户在页面上针对队列名字，可以配置 积压超标、qps骤降、消费者掉线、失败率飙升、平均耗时高 5种告警指标。
### 6.30.6 elk + grafana 实现错误告警
将 Funboost 的日志文件夹通过 Filebeat/Logstash 采集到 Elasticsearch，然后在 Grafana 中配置 Elasticsearch 数据源的ERROR日志比例或者个数，告警规则。这种方式适合已有 ELK 体系的环境。
懂elk和grafana的人都知道怎么做，无需我长篇展开。
### 6.30.10 错误触发警告的方式选择
以上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()` 发送的消息 |
```python
# 假设 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就会越多。
```json
{
"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
}
```
> 💡 **适用场景**：调试消息格式、确认序列化结果、排查消息投递问题。