一、定时任务工具选型
1、几个开原框架
分别从
1） https://github.com/celery/celery
2）https://github.com/agronholm/apscheduler
3）https://github.com/ydf0509/funboost
4）https://github.com/dbader/schedule
最终选择：schedule 框架
二、启动定时任务
您可以使用Python中的schedule库来编排任务，并且通过配置文件的方式管理任务执行时间和Corn表达式。以下是一个示例，演示如何使用schedule库来实现这一目标
安装schedule库（如果尚未安装）：
pip install schedule
1、配置 .ini
创建一个配置文件，例如config.ini，其中包含任务名称、Corn表达式和间隔时间（以秒为单位）：
.ini 文件
; 这里是黑
[TaskA1]
corn = 09:15
method = catch_dayahead_load_forecast
param = -2[TaskA2]
corn = 09:16
method = catch_dayahead_line_forecast
param = -2[TaskA3]
corn = 09:17
method = catch_newenergy_totalpower_forecast
param = -2[TaskA4]
corn = 09:18
method = catch_hydroplan_totalpower_forecast
param = -2上面的三个参数全部是自定义的
我们可以根据需求任意更改属性名，增加、或者删除
- corn：任务调度的时间节点
- method：方法名（无需路径）
- param：单个参数（这里可以通过多属性或者单属性隔离方式处理）
2、定时任务程序
创建一个Python脚本来读取配置文件并设置任务计划：
这里我们使用 python的标准库中的configparser进行.ini文件配置的读取
import configparser
import schedule
import timedef job(task_name):print(f"Executing task: {task_name}")def load_config(file_path):config = configparser.ConfigParser()config.read(file_path)return configdef main():config = load_config('config.ini')for task_name in config.sections():task = config[task_name]corn_expression = task['corn']method_name = task['method']param = task['param']# 根据方法名称获取对应的方法引用method = globals()[method_name]schedule.every().day.at(corn_expression).do(method, param).tag(task_name)while True:schedule.run_pending()time.sleep(1)if __name__ == "__main__":main()
运行上述Python脚本，它将从配置文件中读取任务信息，并使用schedule库来设置Corn表达式和间隔时间，然后按照您的要求执行任务。
这个示例中的任务将会每隔1分钟执行一次，并且可以根据配置文件中的Corn表达式来自定义执行时间。您可以根据需要添加更多的任务到配置文件中，然后运行脚本来执行它们。
3、加载外部程序
当 .ini 配置中使用了一个不在定时任务中的方法
则我们需要在加载定时任务的地方将这个路径加上
如：from clearing_elec_info.dayhead_new_energy_scrapy import dayhead_new_energy_scrapy_elec
否则运行时会报错
import configparser
import sys
import schedule
import timeimport osfrom common.device_list import DAYAHEAD_DEVICE
from common.utils.min_add_func import min_add_func# os.chdir('')
sys.path.insert(0, os.path.join(os.path.dirname(__file__), os.pardir))
# 这里如果更改文件名、目录结构、或者方法名一定记得同步修改这里from clearing_elec_info.dayhead_new_energy_scrapy import dayhead_new_energy_scrapy_elec# 加载配置
def load_config(file_path):config = configparser.ConfigParser()config.read(file_path)return configdef main():config = load_config('schedule_task/config.ini')for task_name in config.sections():task = config[task_name]corn_expression = task['corn']method_name = task['method']param = task['param']# 根据方法名称获取对应的方法引用method = globals()[method_name]print("-------------加载任务完成！等待执行")while True:schedule.run_pending()time.sleep(1)if __name__ == "__main__":main()
