定时任务改成redis

Author: anjia0532

app/scheduler/scheduler.py

@@ -1,10 +1,9 @@
-from apscheduler.triggers.cron import CronTrigger
-from funboost import funboost_aps_scheduler
+from funboost.timing_job.apscheduler_use_redis_store import funboost_background_scheduler_redis_store
 
 from core.lib import logger
 
 LOGGER = logger.for_service('scheduler')
 
 
 def register_scheduler(app):
-    funboost_aps_scheduler.start()
+    funboost_background_scheduler_redis_store.start()

app/tasks/common.py

@@ -6,29 +6,29 @@
 
 class FunboostCommonConfig(BoosterParams):
     # 中间件选型见3.1章节 https://funboost.readthedocs.io/zh/latest/articles/c3.html
-    broker_kind: str = BrokerEnum.SQLITE_QUEUE
+    broker_kind: BrokerEnum = BrokerEnum.REDIS_STREAM
     # 最大自动重试次数
     max_retry_times: int = 4
     # 函数出错后间隔多少秒再重试.
     retry_interval: typing.Union[float, int] = 30
     # 消费者和发布者的日志级别
-    log_level: int = logging.INFO
+    log_level: int = logging.DEBUG
     # 是否使用分布式控频
-    is_using_distributed_frequency_control: bool = False
+    is_using_distributed_frequency_control: bool = True
     # # 函数达到最大重试次数仍然没成功，是否发送到死信队列,死信队列的名字是 队列名字 + _dlx。
     is_push_to_dlx_queue_when_retry_max_times: bool = True
     # 任务过滤的失效期，为0则永久性过滤任务。例如设置过滤过期时间是1800秒 ， 30分钟前发布过1 + 2 的任务，现在仍然执行，如果是30分钟以内发布过这个任务，则不执行1 + 2
     task_filtering_expire_seconds: int = 0
     # 消息过期时间，为0永不过期，为10则代表，10秒之前发布的任务如果现在才轮到消费则丢弃任务。
-    msg_expire_senconds: int = 120
+    msg_expire_senconds: int = 20
     # # 是否对函数入参进行过滤去重.
     do_task_filtering: bool = False
     # 运行时候,是否记录从消息队列获取出来的消息内容
     is_show_message_get_from_broker: bool = True
     # 提供一个用户自定义的保存消息处理记录到某个地方例如mysql数据库的函数，函数仅仅接受一个入参，入参类型是 FunctionResultStatus，用户可以打印参数
     # user_custom_record_process_info_func: typing.Callable = save_result_status_to_sqlalchemy
     # 是否将发布者的心跳发送到redis，有些功能的实现需要统计活跃消费者。因为有的中间件不是真mq。这个功能,需要安装redis.
-    is_send_consumer_hearbeat_to_redis: bool = False
+    is_send_consumer_hearbeat_to_redis: bool = True
     # 是否支持远程任务杀死功能，如果任务数量少，单个任务耗时长，确实需要远程发送命令来杀死正在运行的函数，才设置为true，否则不建议开启此功能。
     is_support_remote_kill_task: bool = False
     function_timeout: typing.Union[int, float] = 600

app/tasks/task_syncer.py

@@ -5,6 +5,7 @@
 
 from apscheduler.triggers.cron import CronTrigger
 from funboost import boost, funboost_aps_scheduler, funboost_current_task
+from funboost.timing_job.apscheduler_use_redis_store import funboost_background_scheduler_redis_store
 from nb_time import NbTime
 
 from app.model.syncer_model import Jobs, DiscoveryInstance, Registration
@@ -51,7 +52,7 @@ def get_gateway_client(name: str) -> Gateway:
 
 
 def clear_client():
-    funboost_aps_scheduler.remove_all_jobs()
+    funboost_background_scheduler_redis_store.remove_all_jobs()
     from app.model.config import discovery_clients, gateway_clients
 
     discovery_clients.clear()
@@ -206,18 +207,18 @@ def reload():
                 Jobs(**target.dict()).save_or_update(sqla_helper)
                 # 注册定时任务
                 if next_run_time:
-                    funboost_aps_scheduler.add_push_job(syncer, id=target.id, name=target.id,
+                    funboost_background_scheduler_redis_store.add_push_job(syncer, id=target.id, name=target.id,
                                                         next_run_time=next_run_time,
                                                         kwargs={"target": target.model_dump()}, replace_existing=True)
                 else:
-                    funboost_aps_scheduler.add_push_job(syncer, id=target.id, name=target.id,
+                    funboost_background_scheduler_redis_store.add_push_job(syncer, id=target.id, name=target.id,
                                                         trigger=CronTrigger(second=second, minute=minute, hour=hour,
                                                                             day=day, month=month,
                                                                             day_of_week=day_of_week),
                                                         kwargs={"target": target.model_dump()}, replace_existing=True)
                 # 健康检查
                 if target.healthcheck:
-                    funboost_aps_scheduler.add_push_job(health_check, id="health-check",
+                    funboost_background_scheduler_redis_store.add_push_job(health_check, id="health-check",
                                                         name="health-check",
                                                         trigger=CronTrigger(second=second, minute=minute, hour=hour,
                                                                             day=day, month=month,