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

news2024/9/19 10:38:24

描述 

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 platform

from 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:
            pass

        def _unlock_file():
            try:
                f.seek(0)
                msvcrt.locking(f.fileno(), msvcrt.LK_UNLCK, 1)
            except Exception as e:
                pass
        atexit.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:

    @classmethod
    def 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.coloradmin.cn/o/2116209.html

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

相关文章

Transformer预测 | 基于Transformer心率时间序列预测(tensorflow)

效果一览 基本介绍 Transformer预测 | 基于Transformer心率时间序列预测(tensorflow) 程序设计 import pandas as pd from pandas.plotting import lag_plot from statsmodels.graphics

暴雨AMD,一起更YES

9月6日,暴雨信息联合AMD举办了“创新驱动 智能未来——2024年行业技术与应用分享会”。在这里,我们与行业领袖、技术专家们一起,深入探讨AI技术的前沿技术动态,洞悉高性能计算的发展趋势,在思维的碰撞和智慧的交融中&a…

BERT 论文逐段精读【论文精读】

BERT: 近 3 年 NLP 最火 CV: 大数据集上的训练好的 NN 模型,提升 CV 任务的性能 —— ImageNet 的 CNN 模型 NLP: BERT 简化了 NLP 任务的训练,提升了 NLP 任务的性能 BERT 如何站在巨人的肩膀上的?使用了哪些 NLP 已有的技术和思想&#xff…

进程等待与退出

目录 前言 0.进程如何退出 1.进程status的获取 2.进程status的组成 3.退出status的分用与验证 4.解释一下errno与waitpid输中status参数的关系 5.status设计 前言 我们在编写程序的时候,我们往往需要知道这个程序中的某个进程执行情况是怎么样的&#xff…

我搞了一台switch

准确的说我搞了一台switch oled。为啥要买个游戏机呢?前些日子去轰趴馆团建,玩了一会switch,当时玩的游戏是《刺客信条3重制版》,感觉挺好玩的。我其实是手残党,很多网络游戏玩的都不是很好,而且大学的时候…

【阿一网络安全】如何让你的密码更安全?(二) - 非对称加密

上次《【阿一网络安全】如何让你的密码更安全?(一) - 对称加密》提到加密算法的对称加密,我们这次来聊聊非对称加密。 和对称加密不同,非对称加密的加密密钥和解密密钥不同。 非对称加密 大概过程就是,发送方使用公钥对明文数据…

Java 后端接口入参 - 联合前端VUE 使用AES完成入参出参加密解密

加密效果&#xff1a; 解密后的数据就是正常数据&#xff1a; 后端&#xff1a;使用的是spring-cloud框架&#xff0c;在gateway模块进行操作 <dependency><groupId>com.google.guava</groupId><artifactId>guava</artifactId><version>30…

Unity | 内存优化之Reserved Unity内存

目录 一、Render Texture资源 1.抗锯齿 2.阴影分辨率 3.深度 4.HDR格式 二、动画资源 1.Resample Curves 2.动画压缩 (1)Keyframe Reduction (2)Optimal(推荐) 3.剔除Scale曲线 (1)开启Remove Constant Scale Curves (2)使用脚本剔除scale曲线 4.降低精…

WordPress上可以内容替换的插件

插件下载地址&#xff1a;WordPress内容替换插件 – 果果开发 类型 替换的类型&#xff1a;文章、自定义文章类型、分类、标签、媒体库、页面、评论、数据库表&#xff0c;不同的类型可以替换不同的字段。 替换字段 替换的字段&#xff0c;哪些字段内容需要替换。除了数据库…

《系统架构设计师教程(第2版)》第17章-通信系统架构设计理论与实践-03-移动通信网网络架构

文章目录 1. 5GS与DN互连1.1 5GS概述1.2 5GS 与DN网络的连接关系1.3 UE连接DN的两种模式1.3.1 透明模式1.3.2 非透明模式 2. 5G 网络边缘计算 1. 5GS与DN互连 1.1 5GS概述 5GS&#xff1a;5G SystemDN&#xff1a;Data NetworkIMS&#xff1a;IP Media Subsystem&#xff08;一…

Android逆向(反调,脱壳,过ssl证书脚本)

文章目录 总结 基础Android基础工具 定位关键代码页面activity定位数据包参数定位堆栈追踪 编写反调脱壳好用的脚本过ssl证书校验抓包反调的脚本打印堆栈bilibili反调的脚本 总结 暑假做了两个月的Android逆向&#xff0c;记录一下自己学到的东西。对于app渗透有了一些思路。 …

Unity下如何播放8K超高分辨率的RTMP流?

在Unity中使用RTMP播放器播放8K流&#xff0c;需要考虑到多个方面的因素和技术要求。以下是一个详细的步骤和要点概述&#xff0c;帮助实现这一目标&#xff1a; 1. 选择合适的RTMP播放器插件 首先&#xff0c;需要选择一个支持8K视频流播放的RTMP播放器插件。并非所有插件都…

MATLAB绘图基础6:MATLAB绘图基础

参考书&#xff1a;《 M A T L A B {\rm MATLAB} MATLAB与学术图表绘制》(关东升)。 6.MATLAB绘图基础 6.1 MATLAB绘图基本流程 % 1.创建图形窗口; % 1.1 创建一个空白图形窗口; figure;% 1.2 创建一个带有指定标题的图形窗口; figure(Name, 图形窗口);% 1.3 创建一个具有指定…

【干货分享】基于SSM的体育场管理系统的开题报告(附源码下载地址)

中秋送好礼 中秋佳节将至&#xff0c;祝福大家中秋快乐&#xff0c;阖家幸福。本期免费分享毕业设计作品&#xff1a;《基于SSM的体育场管理系统》。 基于SSM的体育场管理系统的开题报告 一、课题背景与意义 随着全民健身理念的深入人心&#xff0c;体育场已成为广大师生和…

熟悉Kafka组成模块、Kafka消息提交的方式及优缺点

1. Kafka概念 1.1 Kafka组成模块 Kafka其实是一款基于发布与订阅模式的消息系统&#xff0c;如果按常理来设计&#xff0c;大家是不是把消息发送者的消息直接发送给消息消费者&#xff1f;但Kafka并不是这么设计的&#xff0c;Kafka消息的生产者会对消息进行分类&#xff0c;…

【LVI-SAM】激光雷达点云处理点云帧投影LIO-SAM 之ImageProjection实现细节

【LVI-SAM】激光雷达点云处理点云帧投影LIO-SAM 之ImageProjection实现细节 1. ImageProjection激光雷达点云预处理算法1.0 总结&#xff1a;1.1 功能概述&#xff1a;1.2 算法流程&#xff1a; 2. ImageProjection激光雷达点云预处理算法数学推倒3. ImageProjection激光雷达点…

安卓玩机工具------小米工具箱扩展工具 小米机型功能拓展

小米工具箱扩展版 小米工具箱扩展版 iO_Box_Mi_Ext是由晨钟酱开发的一款适用于小米&#xff08;MIUI&#xff09;、多亲&#xff08;2、2Pro&#xff09;、多看&#xff08;多看电纸书&#xff09;的多功能工具箱。该工具所有功能均可以免root实现&#xff0c;使用前&…

图解TCP三次握手|深度解析|为什么是三次

写在前面 这篇文章我们来讲解析 TCP三次握手。 TCP 报文段 传输控制块TCB&#xff1a;存储了每一个连接中的一些重要信息。比如TCP连接表&#xff0c;指向发送和接收缓冲的指针&#xff0c;指向重传队列的指针&#xff0c;当前的发送和接收序列等等。 我们再来看一下TCP报…

[高级人工智能 开放性调研] 近两年来[2022~2024]人工智能应用进展重要案例介绍

文章目录 [高级人工智能 开放性调研] 近两年来[2022-2024]人工智能应用进展重要案例介绍写在前面1. AIGC1.1 LLM | 大语言模型问答系统式的生成式AI文档解读——KimiChat代码生成——Cursor 1.2 AI绘画\视频生成 | Stable Diffusion | OpenAI SoraStable DiffusionOpenAI Sora …

模拟网络丢包常用方法以及工具

文章目录 背景常用方法代码实现使用方法测试代码 使用网络流量控制工具 常用工具Clumsy 背景 在软件开发过程中&#xff0c;经常需要模拟不同的网络环境来测试应用在不同条件下的表现。 这些模拟可以采用多种方式进行&#xff0c;包括在代码中实现随机丢包、随机延时、乱序&am…