flask python 设置定时任务 flask 周期性执行任务方案

这篇具有很好参考价值的文章主要介绍了flask python 设置定时任务 flask 周期性执行任务方案。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

flask 通常使用 flask_apscheduler 框架设计定时任务,flask_apscheduler 功能很全面,能按设定的时间规则执行任务,可以持久化到各类数据库(mysql,redis,mongodb),实现对定时任务增、删、改、查等操作。

安装

pip3 install flask_apscheduler

1、调用方法

方法一:使用 Config 类配置时间规则

from flask import Flask
from flask_apscheduler import APScheduler

class Config(object):
    # 列表类型,如有需要可以定义多个job
    JOBS = [
        {
            'id': 'job_1',                # 一个标识
            'func': '__main__:job1',     # 指定运行的函数
            'args': (1, 2),              # 传入函数的参数
            'trigger': 'interval',       # 指定 定时任务的类型
            'seconds': 5                # 运行的间隔时间
        }
    ]

    SCHEDULER_API_ENABLED = True

def job1(a, b):                          # 运行的定时任务的函数
    print(str(a) + ' ' + str(b))


if __name__ == '__main__':
    app = Flask(__name__)                 # 实例化flask
    app.config.from_object(Config())      # 为实例化的 flask 引入配置
    scheduler = APScheduler()                  # 实例化 APScheduler
    scheduler.init_app(app)                    # 把任务列表放入 flask
    scheduler.start()                          # 启动任务列表
    app.debug = True
    app.run(host='0.0.0.0',port=8000)          # 启动 flask

方法二:使用装饰器

from flask import Flask
from flask_apscheduler import APScheduler


# 实例化 APScheduler
scheduler = APScheduler()

@scheduler.task('interval', id='job_1', args=(1,2),seconds=5)
def job1(a, b):  # 运行的定时任务的函数
    print(str(a) + ' ' + str(b))


if __name__ == '__main__':
    app = Flask(__name__)  # 实例化flask
    scheduler.start()  # 启动任务列表
    app.debug=True
    app.run(host='0.0.0.0',port= 8000)  # 启动 flask

方法三:通过调用 flask_apscheduler 的 api (推荐)

from flask import Flask
from apscheduler.schedulers.background import BackgroundScheduler

# 调度器在后台线程中运行,不会阻塞当前线程
scheduler = BackgroundScheduler()

def job1(a, b):                          # 运行的定时任务的函数
    print(str(a) + ' ' + str(b))

    
scheduler.add_job(func=job1, args=("1","2"),id="job_1", trigger="interval", seconds=5, replace_existing=False)
'''
func:定时任务执行的函数名称。
args:任务执行函数的位置参数,若无参数可不填
id:任务id,唯一标识,修改,删除均以任务id作为标识
trigger:触发器类型,参数可选:date、interval、cron
replace_existing:将任务持久化至数据库中时,此参数必须添加,值为True。并且id值必须有。不然当程序重新启动时,任务会被重复添加。
'''

if __name__ == '__main__':
    app = Flask(__name__)  # 实例化flask
    scheduler.start()  # 启动任务列表
    app.debug=True
    app.run(host='0.0.0.0',port= 8000)  # 启动 flask

实例对象 scheduler 拥有增、删、改、查等基本用法:文章来源地址https://www.toymoban.com/news/detail-811626.html

新增任务:add_job()
编辑任务:modify_job()
删除任务:remove_job(id)(删除所有任务:remove_all_jobs()
查询任务:get_job(id)(查询所有任务:get_jobs()
暂停任务:pause_job(id)
恢复任务:resume_job(id)
运行任务:run_job(id)(立即运行,无视任务设置的时间规则)

到了这里,关于flask python 设置定时任务 flask 周期性执行任务方案的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • flink生成水位线记录方式--周期性水位线生成器

    在flink基于事件的时间处理中,水位线记录的生成是一个很重要的环节,本文就来记录下几种水位线记录的生成方式的其中一种:周期性水位线生成器 1.1 BoundedOutOfOrdernessTimeStampExtractor 他会接收一个表示最大延迟的参数,比如1分钟,意味着如果到达的元素的事件时间和之前到

    2024年02月07日
    浏览(42)
  • Abaqus CAE 2018插件使用详解:基于周期性边界条件定义3D几何模型的实践指南**

    注:这篇文章是为了帮助读者更好地理解和使用Abaqus CAE 2018的插件来定义周期性边界条件。所提供的信息是基于我个人的理解和实践,如有不准确或者有更好的建议,欢迎读者们指正和交流。 一、 引言 在进行无限或半无限域建模时,周期性边界条件可为我们提供了一种模拟

    2024年02月11日
    浏览(35)
  • 【正点原子STM32】RTC实时时钟(RTC方案、BCD码、时间戳、RTC相关寄存器和HAL库驱动、RTC基本配置步骤、RTC基本驱动步骤、时间设置和读取、RTC闹钟配置和RTC周期性自动唤醒配置)

    一、RTC简介 二、STM32 RTC框图介绍 2.1、STM32 F1 RTC结构框图 2.2、STM32 F4 / F7 / H7 RTC结构框图 三、RTC相关寄存器介绍 3.1、RTC基本配置步骤 3.2、RTC相关寄存器(F1) 3.3、RTC相关寄存器(F4 / F7 / H7) 四、RTC相关HAL库驱动介绍 4.1、RTC相关HAL库驱动(F1) 4.2、RTC相关HAL库驱动(F4 / F7 /

    2024年03月27日
    浏览(56)
  • 使用Dream3D和MATLAB从综合构建微结构到创建具有周期性边界条件的Abaqus输入文件的一站式解决方案

    声明 :本文中的所有内容仅供学术研究和讨论,不保证完全无误。对于使用本文内容可能产生的任何后果,作者不承担任何责任。希望大家在使用时,结合自己的实际情况进行酌情调整。 当我们面临材料力学问题,包括材料的疲劳、断裂和塑性等行为的仿真时,一个常见的

    2024年02月10日
    浏览(51)
  • 为什么RIP使用UDP,OSPF使用IP,而BGP使用TCP?为什么RIP周期性地和邻站交换路由信息而BGP却不这样做?

    RIP只和邻站交换信息,使用UDP无可靠保障,但开销小,可以满足RIP要求; OSPF使用可靠的洪泛法,直接使用IP,灵活、开销小; BGP需要交换整个路由表和更新信息,TCP提供可靠交付以减少带宽消耗; RIP使用不保证可靠交付的UDP,因此必须不断地(周期性地)和邻站交换信息才

    2024年02月02日
    浏览(43)
  • SpringBoot 动态操作定时任务(启动、停止、修改执行周期)增强版

    前段时间编写了一篇博客 SpringBoot 动态操作定时任务(启动、停止、修改执行周期 ,该篇博客还是帮助了很多同学。 但是该篇博客中的方法有些不足的地方: 只能通过前端控制器controller手动注册任务。【具体的应该是我们提前配置好我们的任务,配置完成后让springboot应用

    2024年02月13日
    浏览(31)
  • 【openWrt】设置执行定时任务

    遇到一个问题,使用openWrt软路由搭建服务器,在docker装了一个maccmsV10,需要每天执行cj信息定时任务,但是maccmsV10本身不支持执行定时任务的配置的。  看了下,openWrt是支持本身是linux系统,所以是可以设置定时任务的。只需要装了cron服务。 设置计划任务,保存 重启cron服务

    2024年02月16日
    浏览(58)
  • linux服务器ubuntu定时任务cron设置每秒执行

    后端的同学经常会遇到在服务器做一些定时脚本处理,比如定时清理日志,定时请求某个接口做特殊的业务处理,大多数linux版本都会带cron服务,以ubuntu服务器做个示例。 cron 是一个在 Unix-like 操作系统中用于定时执行任务的工具。它是 “Chronos”(希腊语,意为时间)的缩写

    2024年02月02日
    浏览(46)
  • python 定时任务执行命令行

    1.使用场景: 定时执行jmeter脚本,通过python定时器隔一段时间执行命令行命令。 2.库: os、datetime、threading (1)利用threading.Timer()定时器实现定时任务 Timer方法 说明 Timer(interval, function, args=None, kwargs=None) 创建定时器 cancel() 取消定时器 start() 使用线程方式执行 join(self, timeout

    2023年04月18日
    浏览(28)
  • 在linux上启动、重启、查询、停止java服务脚本,并且设置定时任务自动执行

    以下代码来源于另一位博主,在实践过程中需要设置定时任务,所以遇到一些问题,把没有写清楚的地方优化了一下。 我们把上面的脚本复制到以下目录的脚本文件/data/www/shtools/startmanage.sh,执行命令可以得到不同的结果。 设置的三个定时任务运行脚本, 第一个是0 3 * * *

    2024年02月07日
    浏览(47)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包