说明

我想使用flask_apscheduler进行动态的定时任务调度，这样可以搭建一个建的分布式任务处理系统。

内容

这篇文章总结的很好，使用apscheduler常见的三种方式：

• 1 基于flask,在视图函数中进行任务管理（本篇采用这种）
• 2 基于flask,在配置项中管理。（网上其他常见的）
• 3 直接使用apscheduler启动，比较适合本地的worker，任务相对固定。

采用第一种(flask + 视图函数)的原因：

• 1 使用网页的形式管理任务更方便（增加、暂定、删除）
• 2 观察任务的执行情况比较方便
• 3 观察任务的结果也比较方便

使用flask的调度任务可以视为是偏IO型的，对于复杂的计算任务flask的调度只是把这个任务存到任务数据库中（Mongo)，然后由其他的Worker去执行。

Worker的启动可以采用第三种方式，启动和管理都相对简单。而且Worker本身并不需要分配端口资源，因此很容易用docker-compose进行scale。

1 搭建Flask服务

这部分已经提了很多次，就不重复了

2 任务函数包

建立tasks函数包

3 几种任务模式

• 1 在指定时间执行一次
• 2 在指定时间按一定间隔最多执行n次或者ttl超限
• 3 从指定时间开始无限循环

4 定义任务的参数

4.1 触发 trigger

可以参考这篇文章

4.1.1 只运行一次 date

例如run_date ='2021-01-01 00:00:00'

4.1.2 间隔触发 interval

通过start_date和end_date来约束开始和结束的时间。

4.1.3 cron触发

既可以一次，也可以多次

5 对应的几个设置点

5.1 初始化设置 __init__.py

...
# >>>>>>> 定时任务
from flask_apscheduler import APScheduler 

scheduler = APScheduler()
...

# >>>>>>>>>>>>>>>>>>> 创建app的函数，可以用于以后灵活的加插件
def create_app(config_name):
    app = Flask(__name__)
    app.config.from_object(config[config_name])
    config[config_name].init_app(app)

	...
    scheduler.init_app(app=app)
    scheduler.start()
	...
    return app 

5.2 在static下增加tasks函数包

将函数包放在这里，主要是方便未来直接通过flask增加任务

static
├── img
├── js
├── mds
├── tasks
│   ├── __init__.py
│   ├── task1.py
│   └── task2.py
└── vendor

5.3 对应的几个任务控制

查看当前任务列表、新增任务、暂停任务、继续任务和删除任务。
简单列一下新增和暂停任务的视图函数，其他几个的类似（换一句命令而已）

5.3.1 新增

看起来比较长的原因是我按照自定义的格式去规范视图函数的几个处理。另外cron任务的新增我没写，因为打算还是把task包成一个对象比较好。

from . import app2
from . import dm
from flask import jsonify, request, current_app
from . import scheduler
from . import dynamic_load_func
from . import itasks



# 增加一个任务
@app2.route('/view2/', methods = ['GET', 'POST'])
def view2():
    web1 = dm.WebMsg('view2')

    conf = current_app.config

    input_para_dict = {
    }
    input_para_dict['req_handler'] = request

    # 1 输入步结果
    input_result = before_view2(input_para_dict)

    if not input_result['status']:
        web1.status = False 
        web1.msg ='输入步失败'
        print(input_result)
        return jsonify(web1.to_dict())
    # 2 处理步
    process_dict = input_result['data']
    process_para_dict = {
    }
    process_para_dict['scheduler'] = scheduler
    process_para_dict['itasks'] = itasks
    process_para_dict['the_mongo'] =  conf['THE_MONGO']

    process_result = process_view2(process_dict, process_para_dict)
    if not process_result['status']:
        web1.status = False 
        web1.msg = '处理步失败'
        print(process_result)
        return jsonify(web1.to_dict())

    web1.status = True 
    web1.msg ='ok'
    web1.data = process_result['data']
    return jsonify(web1.to_dict())




# 前置函数
def before_view2(input_para_dict = None):
    web1 = dm.WebMsg('before_view2')

    # 1 获取参数
    try:
        req_handler = input_para_dict['req_handler']
    except:
        web1.status = False 
        web1.msg ='获取request handler失败'
        return web1.to_dict()
    # 2 获取数据
    try:
        input_dict = req_handler.get_json()
        # print(input_dict)
    except:
        web1.status = False 
        web1.msg = '获取前端的数据失败'
        return web1.to_dict()
    
    # 3 处理
    try:
        process_dict = {
    }
        # 任务的名称（id） task1,task2...
        process_dict['task_id'] = input_dict['task_id']
        process_dict['task_name'] = input_dict['task_name']
        process_dict['task_type'] = input_dict['task_type']
        # 任务的参数
        process_dict['task_args']= input_dict.get('task_args')
        process_dict['task_kwargs']= input_dict.get('task_kwargs')
        # 开始和结束时间
        process_dict['task_start_date_str']= input_dict.get('task_start_date_str')
        process_dict['task_end_date_str']= input_dict.get('task_end_date_str')
        # interval风格的参数
        process_dict['weeks']= input_dict.get('weeks')
        process_dict['days']= input_dict.get('days')
        process_dict['hours']= input_dict.get('hours')
        process_dict['minutes']= input_dict.get('minutes')
        process_dict['seconds']= input_dict.get('seconds')
        # cron风格的参数
        process_dict['year']= input_dict.get('year')
        process_dict['month']= input_dict.get('month')
        process_dict['day']= input_dict.get('day')
        process_dict['week']= input_dict.get('week')
        process_dict['day_of_week']= input_dict.get('day_of_week')
        process_dict['hour']= input_dict.get('hour')
        process_dict['minute']= input_dict.get('minute')
        process_dict['second']= input_dict.get('second')
    except:
        web1.status = False
        web1.msg = '整合输入数据失败'
        return web1.to_dict()

    web1.status = True 
    web1.msg = 'ok'
    web1.data = process_dict
    return web1.to_dict()



# 处理函数
def process_view2(process_dict = None,process_para_dict = None):
    web1 = dm.WebMsg('process_view2')
    # 1 获取数据库连接
    try:
        the_mongo = process_para_dict['the_mongo']
    except:
        web1.status = False 
        web1.msg ='获取pymongo连接对象失败'
        return web1.to_dict()

    try:
        scheduler = process_para_dict['scheduler']
    except:
        web1.status = False 
        web1.msg ='获取scheduler对象失败'
        return web1.to_dict()

    try:
        itasks = process_para_dict['itasks']
    except:
        web1.status = False 
        web1.msg ='获取itasks对象失败'
        return web1.to_dict()

    # 2 获取查询参数
    try:
        task_type = process_dict['task_type']
        task_id = process_dict['task_id']
        # 一个task_name就是一个函数，可以起多个task(也就是有多个task_id)
        task_name = process_dict['task_name']
        task_start_date_str = process_dict['task_start_date_str']
        task_args = process_dict['task_args']
        task_kwargs = process_dict['task_kwargs']
        if task_type =='date':
            print('a')
            the_func = dynamic_load_func(itasks,task_name)
            scheduler.add_job(func=the_func, id=task_id,trigger='date', 
                            run_date=task_start_date_str, args =task_args,
                            kwargs =task_kwargs)
        elif task_type=='interval':
            print('b')
            seconds = process_dict['seconds']

            the_func = dynamic_load_func(itasks,task_name)
            print(the_func)
            # the_func()
            scheduler.add_job(func=the_func, id=task_id, trigger=task_type, 
                               seconds= seconds, args = task_args, kwargs = task_kwargs)
    except:
        web1.status = False 
        web1.msg='添加任务失败'
        return web1.to_dict()
    web1.status = True 
    web1.msg = '发布成功'
    web1.data['task_id'] = task_id
    return web1.to_dict()

5.3.2 暂停

主要的函数是scheduler.pause_job(task_id)

...
    # 2 处理步 - 这步目前也是灵活的
    try:
        task_id = process_dict['task_id']
        task_name = process_dict['task_name']

        # 暂停
        scheduler.pause_job(task_id)
    
    except:
        web1.status = False 
        web1.msg='暂停任务失败' 
        web1.data['task_id'] = task_id
        return web1.to_dict()
...

6 Next

任务的几种操作测试下来都是没问题的，但是这样写代码太冗长了，因此之后打算写一个task对象来封装这些操作。