Python Flask_APScheduler定时任务的正确(最佳)使用

描述 

APScheduler基于Quartz的一个Python定时任务框架,实现了Quartz的所有功能。最近使用Flask框架使用Flask_APScheduler来做定时任务,在使用过程当中也遇到很多问题,例如在定时任务调用的方法中需要用到flask的app.app_context()时,需要使用current_app记录日志时,例如:current_app.logger.info("my_job已执行"),定时任务中使用current_app对象会报错,查看了很多资料,大部分资料都是说没有app就创建一个,这样确实也能解决,但是我总感觉这种解决是有问题的,拿.Net Core来说,使用Quartz.NET定时任务时,定时任务依赖于一个Host(主机)对象,不需要重复创建Host对象,但是Flask的app对象使用过程中却需要重新create app,Quartz.NET也是基于Quartz的定时任务框架,我使用过Quartz.NET,因此始终觉得Flask_APScheduler中create app使用是有问题,终于在过了一段时间后看到一位前辈使用Flask_APScheduler的一篇文章后,瞬间通达了,这个问题终于得到完美解决

 最佳使用Flask_APScheduler

 安装Flask_APScheduler

pip install Flask_APScheduler

1.项目结构图如下:

 

 2.Python 软件包utils下的__init__.py 初始化生成APScheduler对象

这里可以灵活处理,例如:也可以是common软件包下__init__.py里初始化APScheduler

 __init__.py的代码如下:

import atexit
import platformfrom flask_apscheduler import APScheduler# 初始化生成APScheduler对象
scheduler = APScheduler()def init_scheduler(app):# 解决APScheduler定时任务重复执行的问题if platform.system() == 'Linux':# Linux 环境下fcntl = __import__("fcntl")f = open('scheduler.lock', 'wb')try:fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)scheduler.init_app(app)scheduler.start()except Exception as e:app.logger.error(e)print(e)def unlock():fcntl.flock(f, fcntl.LOCK_UN)f.close()atexit.register(unlock)else:# Window 环境下msvcrt = __import__('msvcrt')f = open('scheduler.lock', 'wb')try:msvcrt.locking(f.fileno(), msvcrt.LK_NBLCK, 1)scheduler.init_app(app)scheduler.start()except Exception as e:passdef _unlock_file():try:f.seek(0)msvcrt.locking(f.fileno(), msvcrt.LK_UNLCK, 1)except Exception as e:passatexit.register(_unlock_file)

3. config.py配置类代码

class Config:JOBS = [{'id': 'job1','func': 'app:MyService.my_job',  # 注意这里的格式,app 是 Flask 应用对象的名称(app.py),: 后面是任务函数名'kwargs': {'job_name': 'job1'},'trigger': 'cron','hour': 16,  # 16 点执行'minute': 58,  # 58 分执行'second': 0  # 0 秒执行},{'id': 'job2','func': 'app:MyService.my_job',  # 注意这里的格式,app 是 Flask 应用对象的名称(app.py),: 后面是任务函数名'kwargs': {'job_name': 'job2'},'trigger': 'cron','hour': 16,  # 16 点执行'minute': 58,  # 58 分执行'second': 3  # 3 秒执行},{'id': 'job3','func': 'app:MyService.my_job',  # 注意这里的格式,app 是 Flask 应用对象的名称(app.py),: 后面是任务函数名'kwargs': {'job_name': 'job3'},'trigger': 'cron','hour': 16,  # 16 点执行'minute': 58,  # 58 分执行'second': 6  # 6 秒执行}]# 开启API功能,这样才可以用api的方式去查看和修改定时任务SCHEDULER_API_ENABLED = True

4.app.py中代码如下 

from config.config import Config  # 导入Config类的配置
from utils import init_scheduler  # 导入init_scheduler方法
# 创建Flask应用
app = Flask(__name__)
app.config.from_object(Config)  # 读取Config类的配置
init_scheduler(app)  # init_scheduler方法

5. MyService类中的my_job的方法使用app上下文

from flask import current_app  # 导入flask的current_app(当前app)
from utils import scheduler  # 很关键的一步 导入utils.__init__.py 初始化后的scheduler对象class MyService:@classmethoddef my_job(cls, job_name):# # # 此方法在定时任务多的情况下,会有性能问题,少的情况没啥问题# app = create_app()# with app.app_context():#     current_app.logger.info("my_job已执行")# #     print(f"my_job,当前时间{datetime.now()}")# # 使用全局APP变量# get_app()# with APP.app_context():#     current_app.logger.info(f"{job_name}已执行")#     print(f"my_job,当前时间{datetime.now()}")with scheduler.app.app_context():   # 这个sheduler是带有app及其上下文的current_app.logger.info(f"{job_name}已执行")

不建议使用 创建一个app的方法

create app的链接:https://blog.csdn.net/weixin_41934979/article/details/140406152 

6.执行效果如下: 

 源代码地址:https://gitee.com/jxzcode_admin/flask-project.git

 7.总结

 使用的Python 软件包下的__init__.py文件中初始化生成scheduler对象,此对象项目启动后只生成一次,然后导入scheduler对象,在定时任务执行的方法使用: with scheduler.app.app_context(): 就可以 获取flask当前app上下文,不需要create app,个人觉得这才是真正正确使用Flask_APScheduler

 参考资料

https://blog.csdn.net/arnolan/article/details/84936075

https://www.jianshu.com/p/d5a46b2d2fd3

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/bicheng/53665.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

无影云电脑:在最破的电脑上玩最顶配的游戏

关注卢松松,会经常给你分享一些我的经验和观点 我对云电脑很感兴趣,这几天我深度体验了无影云电脑的个人版.,我给大家分享下。这款云电脑到底能不能替代你的笔记本?到底能不能改变人们使用电脑的方式? 先说结论: (1)从草根创…

【Canvas与艺术】菊花孔雀螺旋

【成图】 【代码】 <!DOCTYPE html> <html lang"utf-8"> <meta http-equiv"Content-Type" content"text/html; charsetutf-8"/> <head><title>菊花孔雀螺旋</title><style type"text/css">…

vue3实现打飞机(雷电)

代码可直接运行直接玩&#xff0c;而且要自己加上一些随机事件都很简单了&#xff08;例如发射速度变快&#xff0c;子弹变大&#xff0c;敌人变慢等&#xff09; <template><div class"flex items-center justify-center h-100vh w-full"><div>S…

.net MAUI应用生命周期

.NET Multi-platform App UI (.NET MAUI) 应用通常有四种执行状态&#xff1a;“未运行”、“运行中”、“已停用”和“已停止”。 当应用从未运行状态转换为运行状态、从运行状态转换为已停用状态、从已停用状态转换为已停止状态、从已停止状态转换为运行状态&#xff0c;以及…

【Kubernetes】K8s 的鉴权管理(二):基于属性 / 节点 / Webhook 的访问控制

K8s 的鉴权管理&#xff08;二&#xff09;&#xff1a;基于属性 / 节点 / Webhook 的访问控制 1.基于属性的访问控制&#xff08;ABAC 鉴权&#xff09;2.基于节点的访问控制&#xff08;node 鉴权&#xff09;2.1 读取操作2.2 写入操作 3.基于 Webhook 的访问控制3.1 基于 We…

替换cython_bbox库中bbox_ious

说明一下问题&#xff1a;目标追踪代码里往往用到cython_bbox中的bbox_ious。但是该库需要用到 VC&#xff0c;按照有些麻烦。于是采用直接替换该方法&#xff0c;用纯代码实现&#xff0c;无需调用库。 File “src\cython_bbox.pyx”, line 17, in cython_bbox.bbox_overlaps…

深度学习-01 Pytorch

torchvision是一个用于计算机视觉任务的Python包&#xff0c;它是PyTorch的一个扩展库。它提供了一些流行的数据集、模型架构和图像转换函数&#xff0c;以方便用户进行计算机视觉任务的开发和研究。 1.torchvision中包含了许多常用的计算机视觉数据集&#xff0c;如MNIST、CIF…

【AcWing】861. 二分图的最大匹配(匈牙利算法)

匈牙利算法&#xff0c;他可以在比较快的时间复杂度之内告诉我们左边和右边成功匹配的最大数是多少 匹配指的是边的数量&#xff0c;成功的匹配指的是两个未被使用的点之间存在一条边(就不存在两条边共用了一个点的)。 匈牙利算法可以返回成功匹配的最大匹配数是多少。 #incl…

四、搭建网站服务器超详细步骤——解决宝塔界面无法登录问题

前言 本篇博客是搭建网站服务器的第四期&#xff0c;也到了中间的一节 先分享一下我在搭建网站时的个人感受&#xff0c;我在这个环节卡住了很久 后来突然醒悟了&#xff0c;然后成功进入了宝塔界面 现在就来分享一下&#xff0c;我所遇到的问题 小伙伴们坐好了 …

MySQL8 windows archive 安装

MySQL8超详细安装教程&#xff08;MySQL8安装图文教程&#xff09;_MySQL8安装包-CSDN博客 【补充&#xff1a;】 1.mysqld --install mysql // mysql是本次安装的服务命名&#xff0c;也可以叫做mysql8 etc. 2.手动卸载 2.1.net stop mysql 2.2.删除datadir // 即my.ini配置…

项目实战 ---- 商用落地视频搜索系统(9)---UI与上层service的交互优化

目录 背景 第一次优化(UI优化) 优化前UI 优化方向与问题 代码 修改效果 第二次优化(整合优化) 优化方向与问题 代码 优化效果 第三次优化(js层优化) 优化方向与问题 代码 优化效果 第四次优化(UI逻辑再优化) 优化方向与问题 代码 优化效果 第五次优化(纯U…

通信工程学习:什么是FMC固定移动融合

FMC&#xff1a;固定移动融合 FMC固定移动融合&#xff0c;即Fixed Mobile Convergence&#xff08;固定移动网络融合&#xff09;&#xff0c;是指通过固定网络与移动网络之间的融通、合作&#xff0c;实现全业务及融合业务的经营。这一技术旨在打破传统固定网络和移动网络之间…

使用 Parallel 类进行多线程编码(上)

用 C# 进行多线程编程有很多方式&#xff0c;比如使用 Thread 对象开启一个新线程&#xff0c;但这已经是一种落后的写法了&#xff0c;现在推荐的写法是使用 Parallel 类&#xff0c;它可以让我们像写传统代码一样编写多线程的程序&#xff0c;Parallel 类有三个常用的方法如下…

C#笔记8 线程是什么?多线程怎么实现和操作?

这和前面的学习内容可能有点不太连贯&#xff0c;但是呢我们一般来说的学习就是遇到什么困难就去学习什么&#xff0c;这也是为什么看那些循序渐进的教程虽然学的很饱满&#xff0c;但是我们有时会学了前面忘记了后面&#xff0c;或者对某个板块理解不深&#xff0c;乃至于写代…

VS Code 中提升编程效率的功能及使用方法

Visual Studio Code&#xff08;简称 VS Code&#xff09;是一款功能强大且灵活的源代码编辑器&#xff0c;支持多种编程语言&#xff0c;并提供了丰富的插件生态系统。对于开发者来说&#xff0c;掌握一些提升编程效率的功能是非常重要的。本文将介绍 VS Code 中一些能够显著提…

大学物理实验-杨氏双缝干涉实验

一、实验目的 1.理解杨氏双缝干涉现象的基本原理&#xff1b; 2.掌握用光具座调整杨氏双缝干涉装置的光路&#xff1b; 3.用CCD观察杨氏双缝干涉的实验现象&#xff1b; 4.学会用干涉法测量未知光波的波长。 二、仪器用具 FB760-6型光学实验仪多功能光学片 三、实验原理 1.波…

智能提取:OfficeImagesExtractor让文档图片提取更简单

“科技是国之利器&#xff0c;也是民之福祉。” 在数字化办公日益普及的今天&#xff0c;我们对文档处理的需求也在不断增长。尤其是对于Office文档中的图片、视频和音频等多媒体内容的提取&#xff0c;传统的方法是繁琐且效率低下的。在这样的背景下&#xff0c;一款能够高效、…

计算机毕业设计选题推荐-产品委托配送系统-Java/Python项目实战

✨作者主页&#xff1a;IT毕设梦工厂✨ 个人简介&#xff1a;曾从事计算机专业培训教学&#xff0c;擅长Java、Python、微信小程序、Golang、安卓Android等项目实战。接项目定制开发、代码讲解、答辩教学、文档编写、降重等。 ☑文末获取源码☑ 精彩专栏推荐⬇⬇⬇ Java项目 Py…

视频监控管理平台LntonAIServer视频智能分析噪声检测应用场景

在视频监控系统中&#xff0c;噪声问题常常影响到视频画面的清晰度和可用性。噪声可能由多种因素引起&#xff0c;包括但不限于低光环境、摄像机传感器灵敏度过高、编码压缩失真等。LntonAIServer通过引入噪声检测功能&#xff0c;旨在帮助用户及时发现并解决视频流中的噪声问题…

原点安全荣获“AutoSec Awards 安全之星”优秀汽车数据安全合规方案奖

9月3日&#xff0c;「AutoSec 2024第八届中国汽车网络安全周暨第五届智能汽车数据安全展」在上海盛大开幕。本届大会由谈思实验室和谈思汽车主办、上海市车联网协会联合主办&#xff0c;以汽车“网络数据安全、软件安全、功能安全”为主题&#xff0c;汇聚了国内外的技术专家、…