Python-flask任务调度模块【定时器】 – APScheduler，Flask-APScheduler实现定时任务【内部定时器调用,无需api】

小鱼儿 2021-09-30 10:34 482阅读 0赞

### **有一个注意事项：如果是秒参数为seconds【官方文档和其他的文档都是second是错误的，切记】** ###

官网：[https://apscheduler.readthedocs.io/en/latest/][https_apscheduler.readthedocs.io_en_latest]

APScheduler参数参考文章：[https://www.cnblogs.com/huchong/p/9088611.html\#\_lab2\_6\_1][https_www.cnblogs.com_huchong_p_9088611.html_lab2_6_1]

Flask-APScheduler 爬坑指南：[https://www.jianshu.com/p/2628f566b31c][https_www.jianshu.com_p_2628f566b31c]

官方GitHub上的案例：[https://github.com/viniciuschiele/flask-apscheduler/tree/master/examples][https_github.com_viniciuschiele_flask-apscheduler_tree_master_examples]

### **个人配置成功案例总结：** ###

### **\_\_init\_\_.py文件安装插件到app上** ###

# 注册APScheduler调度器【内部定时器,无需api调用】
        def __app_register_APScheduler(self):
            from flask_apscheduler import APScheduler
            from app.libs.apscheduler.OwnAPScheduler import OwnAPScheduler
            # 激活APScheduler调度器【内部定时器必须的参数】
            self.app.config["SCHEDULER_API_ENABLED"] = True
            apscheduler = APScheduler()
            apscheduler.init_app(app=self.app)
            OwnAPScheduler(self.app, apscheduler)

### **创建一个OwnAPScheduler类，管理执行job任务【我这里没有使用配置文件,而是自己add\_job方法添加任务】** ###

'''
        当前文件的代码不会被api访问,而是在程序初始化的时候就会定时器调用【必须确保使用aspcheduler对象的时候已经安装到app上】
        【非常重要】api调用(app初始化完毕后才会执行),定时器调度(app初始化过程中会执行),一定要区分执行顺序问题
    '''
    from app.libs.bonus.issue_token_bonus import IssueTokenBonus
    from app.models.bonus.issue_token_model import IssueTokenModel
    
    
    class OwnAPScheduler:
        # 【非常重要】这里不能使用current_app,因为这里是在初始化的时候定时器调度,此刻current_app还没创建
        def __init__(self, app, apscheduler):
            self.app = app
            self.apscheduler = apscheduler
            self.scheduler_all()
    
        # 需要调度的一个job
        def job_issue_token(self):
            # 这里必须推入app的上下文才可以执行！
            with self.app.app_context():
                print("开始执行发行token,每隔1小时")
                IssueTokenModel.start_issue_token_execute()
                IssueTokenBonus.start_issue_token_bonus_execute()
    
        # 添加调度任务且执行
        def scheduler_all(self):
            # add_job参数说明参考：https://www.cnblogs.com/huchong/p/9088611.html#_lab2_6_1
            self.apscheduler.add_job(id="job-issue-token-scheduler-id", func=self.job_issue_token, trigger="interval",
                                     hours=2)
            self.apscheduler.start()

[https_apscheduler.readthedocs.io_en_latest]: https://apscheduler.readthedocs.io/en/latest/
[https_www.cnblogs.com_huchong_p_9088611.html_lab2_6_1]: https://www.cnblogs.com/huchong/p/9088611.html#_lab2_6_1
[https_www.jianshu.com_p_2628f566b31c]: https://www.jianshu.com/p/2628f566b31c
[https_github.com_viniciuschiele_flask-apscheduler_tree_master_examples]: https://github.com/viniciuschiele/flask-apscheduler/tree/master/examples