【python】任务调度编排工具 schedule | python定时任务工具

一、定时任务工具选型

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

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

上面的三个参数全部是自定义的
我们可以根据需求任意更改属性名，增加、或者删除

• corn：任务调度的时间节点
• method：方法名（无需路径）
• param：单个参数（这里可以通过多属性或者单属性隔离方式处理）

2、定时任务程序

创建一个Python脚本来读取配置文件并设置任务计划：
这里我们使用 python的标准库中的configparser进行.ini文件配置的读取

import configparser
import schedule
import time

def job(task_name):
    print(f"Executing task: {task_name}")

def load_config(file_path):
    config = configparser.ConfigParser()
    config.read(file_path)
    return config

def 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()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

运行上述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 time

import os

from 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 config


def 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()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45