最近项目中缺乏一款分布式框架，在github上闲逛时，找到了一款分布式调度框架。按照国惯例先上链接：https://github.com/ydf0509/funboost。该框架是由国人ydf开发的，在最新的16.2版本中，已经支持了20种消息中间键。完全可以满足你不同的开发需求。
如果你是一位使用过celery的开发者，那么你一定知道这个框架能干什么，上手也很快。对于那些没有怎么使用过celery的开发者们，这绝对是一款能够快速上手，并且不用经历celery写一大堆配置文件和陡峭的学习曲线的框架，框架中自带的20种消息中间键中绝对有你熟悉的一款，再也不用费老鼻子劲去搭建中间键，学习中间键怎么使用，然后再来开发你的应用程序。
这个框架的学习难易度绝对远低于celery，并且框架内部所有的注释都是中文，对于英语不好的兄弟们，只能说再也找不到这么友好的框架了。而且还有中文QQ群，随时可以与作者互动，再也不用像在github上面提个问题，还要等时差，随时随地等候着作者回答，更新补丁了。下面是框架的简单架构图，估计作者的天赋全都用在编程上了，画画水平差一点，也是可以理解的。
image.png
对于新手朋友们，这里还是大概说一下这个框架能干什么吧：
分布式爬虫，你再也不用花大量的时间去学习scrapy了。因为它可以帮你们完成scrapy的分布式爬虫，你的爬虫完全可以requests写到底。爬虫爬取速度慢，再也不是瓶颈。这个框架支持多进程+多线程， 多进程+异步。怎么还是慢？那就分布式呗。直接一个docker镜像，部署到多态机器上。在官方文档上，作者已经对于该框架对比scrapy的优势进行了阐述。还有几个爬虫的例子，值得新上手的兄弟们一试。
分布式计算，例如：数据清洗，ETL，数据分析等。该框架也可以胜任分布式工作。该框架对比spark来说，在计算方面还是有很多无法满足的功能，但是对于简单的计算和数据的清洗，是绝对够用了。也不用费很大的劲去搭建hadoop+spark这套环境了。python的Faust框架也可以做数据清洗等工作，但是它只对kafka支持。而该框架有20多种中间键。
该框架完全可以实现类似Airflow工作流自动化和调度系统，可以设置一套有向无环图，每天定时启动任务，任务失败发送邮件等等。值得一提的是，该框架中的日志模块NB_LOG，可以媲美loguru，从名字上看的出来作者还是比较自恋的。当使用后发现这个日志系统确实NB。只需要设置好日志后，完全可以在任务完成的时候给你的邮箱，钉钉，es，mongo推送日志了。对于python的原生日志系统不太了解的兄弟们，这个绝对值得你们一试。用后的感觉就是，妈妈再也不担心我的日志了。
自动化运维。当然自动化运维方面有点类似于Airflow。因为有很多运维的兄弟们也使用Airflow来做自动化运维。
当然还有其他的应用场景，例如测试等等，等待各位开发。
框架特性
文档里有详细介绍
https://funboost.readthedocs.io/zh/latest/articles/c1.html#id10
新特性
框架更新到16.2以后，更新了一个强大的功能就是可以自定义生产者和消费者，定义完成了以后再通过register_custom_broker注册路由函数，注册到框架内部。再为这个路由自定义一个编号就可以使用了。相当方便。当然也可以通过该功能重写框架原有的任务发布者和消费者。
在我的项目中，我就重写了原框架基于redis任务确认的publisher和consumer。重新注册，其中主要对于任务的状态进行了监控。
import json
import time
from funboost import FunctionResultStatus
from funboost.assist.user_custom_broker_register import register_custom_broker
from funboost.concurrent_pool.async_helper import simple_run_in_executor
from funboost.consumers.base_consumer import _delete_keys_and_return_new_dict
from funboost.consumers.redis_consumer_ack_able import RedisConsumerAckAble
from funboost.utils import decorators, time_util
from funboost.utils.redis_manager import RedisManager
from funboost.utils.redis_manager import RedisMixin as InheritRedisMixin
from funboost.publishers.redis_publisher_simple import RedisPublisher as SimpleRedisPublisher
from conf import settings
from pkg.status.status_code import RedisTaskStatusCodeEnum
redis_expire_time = settings.FuncBoost.redis_expire_time # 设置任务状态的过期时间，随便设置，单位是秒，只要是int就行
class RedisMixin(InheritRedisMixin):
"""
继承了框架里的RedisMixin，导入的时候重命名为InheritRedisMixin，重写的时候又把InheritRedisMixin命名成原框架内部的名字RedisMixin。
这样的骚操作可以避免不必要BUG
"""
@property
@decorators.cached_method_result
def redis_db_task_status(self):
"""
在redis中新建一个库来存放任务的状态
:return:
"""
return RedisManager(host=settings.Redis.REDIS_HOST, port=settings.Redis.REDIS_PORT,
password=settings.Redis.REDIS_PASSWORD, db=settings.Redis.REDIS_TASK_STATUS_DB).get_redis()
class RedisConsumeLatestPublisher(SimpleRedisPublisher):
"""
复写Publisher类，继承了SimpleRedisPublisher
"""
def concrete_realization_of_publish(self, msg: str):
"""
发布任务的函数，当发布任务后，在redis中记录任务已经发布成功
:param msg:
:return:
"""
msg_dict = json.loads(msg)
task_id = msg_dict['extra']['task_id']
# msg_dict['status'] = RedisTaskStatusCodeEnum.PUBLISH.status
with RedisMixin().redis_db_task_status.pipeline() as p:
p.set(task_id, RedisTaskStatusCodeEnum.PUBLISH.code)
p.expire(task_id, settings.FuncBoost.task_status_expire_time)
p.execute()
self.logger.info(f'{task_id} == status:{RedisTaskStatusCodeEnum.PUBLISH.code}')
self.redis_db_frame.lpush(self._queue_name, msg)
class RedisConsumeLatestConsumer(RedisConsumerAckAble):
"""
复写Consumer类，继承了RedisConsumerAckAble,该类提供了任务确认功能
"""
def _run(self, kw: dict, ):
"""
重写 AbstractConsumer._run， 同步运行fun的方法
记录任务开始执行，更新任务状态
记录任务执行成功，更新任务状态
任务执行失败，更新任务状态
:param self:
:param kw:
:return:
"""
self.logger.info(f'_run 开始, kw: {kw}')
task_id = kw['body']['extra']['task_id']
# msg_dict['status'] = RedisTaskStatusCodeEnum.PUBLISH.status
with RedisMixin().redis_db_task_status.pipeline() as p:
p.set(task_id, RedisTaskStatusCodeEnum.TASKSTART.code)
p.execute()
self.logger.info(f'{task_id} == status:{RedisTaskStatusCodeEnum.TASKSTART.code}')
t_start_run_fun = time.time()
max_retry_times = self._get_priority_conf(kw, 'max_retry_times')
current_function_result_status = FunctionResultStatus(self.queue_name, self.consuming_function.__name__,
kw['body'], )
current_retry_times = 0
function_only_params = _delete_keys_and_return_new_dict(kw['body'])
for current_retry_times in range(max_retry_times + 1):
current_function_result_status = self._run_consuming_function_with_confirm_and_retry(kw,
current_retry_times=current_retry_times,
function_result_status=FunctionResultStatus(
self.queue_name,
self.consuming_function.__name__,
kw['body']),
)
if current_function_result_status.success is True or current_retry_times == max_retry_times or current_function_result_status.has_requeue:
with RedisMixin().redis_db_task_status.pipeline() as p:
p.set(task_id, RedisTaskStatusCodeEnum.SUCCESS.code)
p.expire(task_id, settings.FuncBoost.task_status_expire_time)
p.execute()
self.logger.info(f'{task_id} == status:{RedisTaskStatusCodeEnum.SUCCESS.code}')
break
self._result_persistence_helper.save_function_result_to_mongo(current_function_result_status)
self._confirm_consume(kw)
if self._get_priority_conf(kw, 'do_task_filtering'):
self._redis_filter.add_a_value(function_only_params) # 函数执行成功后，添加函数的参数排序后的键值对字符串到set中。
if current_function_result_status.success is False and current_retry_times == max_retry_times:
with RedisMixin().redis_db_task_status.pipeline() as p:
p.set(task_id, RedisTaskStatusCodeEnum.FAIL.code)
p.expire(task_id, settings.FuncBoost.task_status_expire_time)
p.execute()
self.logger.info(f'{task_id} == status:{RedisTaskStatusCodeEnum.FAIL.code}')
self.logger.critical(
f'函数 {self.consuming_function.__name__} 达到最大重试次数 {self._get_priority_conf(kw, "max_retry_times")} 后,仍然失败， 入参是 {function_only_params} ')
if self._get_priority_conf(kw, 'is_using_rpc_mode'):
# print(function_result_status.get_status_dict(without_datetime_obj=
with RedisMixin().redis_db_filter_and_rpc_result.pipeline() as p:
# RedisMixin().redis_db_frame.lpush(kw['body']['extra']['task_id'], json.dumps(function_result_status.get_status_dict(without_datetime_obj=True)))
# RedisMixin().redis_db_frame.expire(kw['body']['extra']['task_id'], 600)
p.lpush(kw['body']['extra']['task_id'],
json.dumps(current_function_result_status.get_status_dict(without_datetime_obj=True)))
p.expire(kw['body']['extra']['task_id'], redis_expire_time)
p.execute()
with self._lock_for_count_execute_task_times_every_unit_time:
self._execute_task_times_every_unit_time += 1
self._consuming_function_cost_time_total_every_unit_time += time.time() - t_start_run_fun
self._last_execute_task_time = time.time()
if time.time() - self._current_time_for_execute_task_times_every_unit_time > self._unit_time_for_count:
avarage_function_spend_time = round(
self._consuming_function_cost_time_total_every_unit_time / self._execute_task_times_every_unit_time,
4)
msg = f'{self._unit_time_for_count} 秒内执行了 {self._execute_task_times_every_unit_time} 次函数 [ {self.consuming_function.__name__} ] ,' \
f'函数平均运行耗时 {avarage_function_spend_time} 秒'
if self._msg_num_in_broker != -1: # 有的中间件无法统计或没实现统计队列剩余数量的，统一返回的是-1，不显示这句话。
# msg += f''' ，预计还需要 {time_util.seconds_to_hour_minute_second(self._msg_num_in_broker * avarage_function_spend_time / active_consumer_num)} 时间 才能执行完成 {self._msg_num_in_broker}个剩余的任务'''
need_time = time_util.seconds_to_hour_minute_second(
self._msg_num_in_broker / (
self._execute_task_times_every_unit_time / self._unit_time_for_count) /
self._distributed_consumer_statistics.active_consumer_num)
msg += f''' ，预计还需要 {need_time}''' + \
f''' 时间 才能执行完成 {self._msg_num_in_broker}个剩余的任务'''
self.logger.info(msg)
self._current_time_for_execute_task_times_every_unit_time = time.time()
self._consuming_function_cost_time_total_every_unit_time = 0
self._execute_task_times_every_unit_time = 0
if self._user_custom_record_process_info_func:
self._user_custom_record_process_info_func(current_function_result_status)
async def _async_run(self, kw: dict, ):
"""
重写 AbstractConsumer._async_run， 异步运行fun的方法
记录任务开始执行，更新任务状态
记录任务执行成功，更新任务状态
任务执行失败，更新任务状态
虽然和上面有点大面积重复相似，这个是为了asyncio模式的，asyncio模式真的和普通同步模式的代码思维和形式区别太大，
框架实现兼容async的消费函数很麻烦复杂，连并发池都要单独写
"""
self.logger.info(f'_async_run 开始, kw: {kw}')
task_id = kw['body']['extra']['task_id']
# msg_dict['status'] = RedisTaskStatusCodeEnum.PUBLISH.status
with RedisMixin().redis_db_task_status.pipeline() as p:
p.set(task_id, RedisTaskStatusCodeEnum.TASKSTART.code)
p.execute()
self.logger.info(f'{task_id} == status:{RedisTaskStatusCodeEnum.TASKSTART.code}')
t_start_run_fun = time.time()
max_retry_times = self._get_priority_conf(kw, 'max_retry_times')
current_function_result_status = FunctionResultStatus(self.queue_name, self.consuming_function.__name__,
kw['body'], )
current_retry_times = 0
function_only_params = _delete_keys_and_return_new_dict(kw['body'])
for current_retry_times in range(max_retry_times + 1):
current_function_result_status = await self._async_run_consuming_function_with_confirm_and_retry(kw,
current_retry_times=current_retry_times,
function_result_status=FunctionResultStatus(
self.queue_name,
self.consuming_function.__name__,
kw[
'body'], ),
)
if current_function_result_status.success is True or current_retry_times == max_retry_times or current_function_result_status.has_requeue:
with RedisMixin().redis_db_task_status.pipeline() as p:
p.set(task_id, RedisTaskStatusCodeEnum.SUCCESS.code)
p.expire(task_id, settings.FuncBoost.task_status_expire_time)
p.execute()
self.logger.info(f'{task_id} == status:{RedisTaskStatusCodeEnum.SUCCESS.code}')
break
# self._result_persistence_helper.save_function_result_to_mongo(function_result_status)
await simple_run_in_executor(self._result_persistence_helper.save_function_result_to_mongo,
current_function_result_status)
await simple_run_in_executor(self._confirm_consume, kw)
if self._get_priority_conf(kw, 'do_task_filtering'):
# self._redis_filter.add_a_value(function_only_params) # 函数执行成功后，添加函数的参数排序后的键值对字符串到set中。
await simple_run_in_executor(self._redis_filter.add_a_value, function_only_params)
if current_function_result_status.success is False and current_retry_times == max_retry_times:
with RedisMixin().redis_db_task_status.pipeline() as p:
p.set(task_id, RedisTaskStatusCodeEnum.FAIL.code)
p.expire(task_id, settings.FuncBoost.task_status_expire_time)
p.execute()
self.logger.info(f'{task_id} == status:{RedisTaskStatusCodeEnum.FAIL.code}')
self.logger.critical(
f'函数 {self.consuming_function.__name__} 达到最大重试次数 {self._get_priority_conf(kw, "max_retry_times")} 后,仍然失败， 入参是 {function_only_params} ')
# self._confirm_consume(kw) # 错得超过指定的次数了，就确认消费了。
if self._get_priority_conf(kw, 'is_using_rpc_mode'):
def push_result():
with RedisMixin().redis_db_filter_and_rpc_result.pipeline() as p:
p.lpush(kw['body']['extra']['task_id'],
json.dumps(current_function_result_status.get_status_dict(without_datetime_obj=True)))
p.expire(kw['body']['extra']['task_id'], redis_expire_time)
p.execute()
await simple_run_in_executor(push_result)
# 异步执行不存在线程并发，不需要加锁。
self._execute_task_times_every_unit_time += 1
self._consuming_function_cost_time_total_every_unit_time += time.time() - t_start_run_fun
self._last_execute_task_time = time.time()
if time.time() - self._current_time_for_execute_task_times_every_unit_time > self._unit_time_for_count:
avarage_function_spend_time = round(
self._consuming_function_cost_time_total_every_unit_time / self._execute_task_times_every_unit_time, 4)
msg = f'{self._unit_time_for_count} 秒内执行了 {self._execute_task_times_every_unit_time} 次函数 [ {self.consuming_function.__name__} ] ,' \
f'函数平均运行耗时 {avarage_function_spend_time} 秒'
if self._msg_num_in_broker != -1:
if self._msg_num_in_broker != -1: # 有的中间件无法统计或没实现统计队列剩余数量的，统一返回的是-1，不显示这句话。
# msg += f''' ，预计还需要 {time_util.seconds_to_hour_minute_second(self._msg_num_in_broker * avarage_function_spend_time / active_consumer_num)} 时间 才能执行完成 {self._msg_num_in_broker}个剩余的任务'''
need_time = time_util.seconds_to_hour_minute_second(
self._msg_num_in_broker / (
self._execute_task_times_every_unit_time / self._unit_time_for_count) /
self._distributed_consumer_statistics.active_consumer_num)
msg += f''' ，预计还需要 {need_time}''' + \
f''' 时间 才能执行完成 {self._msg_num_in_broker}个剩余的任务'''
self.logger.info(msg)
self._current_time_for_execute_task_times_every_unit_time = time.time()
self._consuming_function_cost_time_total_every_unit_time = 0
self._execute_task_times_every_unit_time = 0
if self._user_custom_record_process_info_func:
await self._user_custom_record_process_info_func(current_function_result_status)
# 自定义发布者和消费者关系的编号
BROKER_KIND_REDIS_CONSUME_LATEST = 103
# 将自定义的发布者和消费者注册到框架内
register_custom_broker_ = register_custom_broker(BROKER_KIND_REDIS_CONSUME_LATEST, RedisConsumeLatestPublisher,
RedisConsumeLatestConsumer) # 核心，这就是将自己写的类注册到框架中，框架可以自动使用用户的类，这样用户无需修改框架的源代码了。
对于上面的代码简单的说明：
redis_db_task_status 一个记录任务状态的库
主要重写的类SimpleRedisPublisher和RedisConsumerAckAble，如果需要在执行任务的前后加一些其他功能，完全可以通过重写_run(同步方法)和_async_run（异步方法）来完成。然后将自定义的新类注册到框架中。
说明
- 该框架整体十分完善，完全可以使用到生产环境中。
- 该框架运用了很多设计模式，注释也是中文，值得大家一读。不清楚的还可以与作者交谈。作为一个学习资料也是一个不错的选择
- 该框架里面有很多作者使用的装饰器，完全可以充实兄弟们开发中的弹药库。当然作者github上也有装饰器的项目。
最后感谢funboost作者ydf0509的贡献。兄弟们如果这篇文章对你有帮助，麻烦点点赞。
