Python使用APScheduler实现定时任务

Kate • 2021年9月11日 11:42 • Python • 692 阅读

一、简介

APScheduler是基于Quartz的一个Python定时任务框架。提供了基于日期、固定时间间隔以及crontab类型的任务，并且可以持久化任务。在线文档：https://apscheduler.readthedocs.io/en/latest/userguide.html

二、安装APScheduler

activate python_test
pip install apscheduler

三、基本概念

APScheduler有四大组件：

1、触发器 triggers

触发器包含调度逻辑。每个作业都有自己的触发器，用于确定下一个任务何时运行。除了初始配置之外，触发器是完全无状态的。有三种内建的trigger:

（1）date: 特定的时间点触发

（2）interval: 固定时间间隔触发

（3）cron: 在特定时间周期性地触发

2、任务储存器 job stores

用于存放任务，把任务存放在内存（为默认MemoryJobStore）或数据库中。

3、执行器 executors

执行器是将任务提交到线程池或进程池中运行，当任务完成时，执行器通知调度器触发相应的事件。

4、调度器 schedulers

把上方三个组件作为参数，通过创建调度器实例来运行根据开发需求选择相应的组件，下面是不同的调度器组件：

BlockingScheduler 阻塞式调度器：适用于只跑调度器的程序。

BackgroundScheduler 后台调度器：适用于非阻塞的情况，调度器会在后台独立运行。

AsyncIOScheduler AsyncIO调度器，适用于应用使用AsnycIO的情况。

GeventScheduler Gevent调度器，适用于应用通过Gevent的情况。

TornadoScheduler Tornado调度器，适用于构建Tornado应用。

TwistedScheduler Twisted调度器，适用于构建Twisted应用。

QtScheduler Qt调度器，适用于构建Qt应用。

四、使用步骤

1、新建一个调度器schedulers

2、添加调度任务

3、运行调度任务

五、使用实例

1、触发器date

特定的时间点触发，只执行一次。参数如下：

参数	说明
run_date (datetime 或 str)	作业的运行日期或时间
timezone (datetime.tzinfo 或 str)	指定时区

使用例子：

from datetime import datetime, date  # 时间
from apscheduler.schedulers.blocking import BlockingScheduler  # 定时任务
from demo import hello, myapp  # 自己写的包
from JobStores import job

'''
三种模式:
（1）date: 特定的时间点触发,最基本的一种调度，作业只会执行一次。
（2）interval: 固定时间间隔触发(在某个时间范围内间隔多长时间执行一次)
（3）cron: 在特定时间周期性地触发(cron 和Linux crontab格式兼容，最为强大)
参数说明:
run_date (datetime 或 str): 作业的运行日期或时间
timezone (datetime.tzinfo 或 str): 指定时区
'''

scheduler = BlockingScheduler()

# 在 2021-9-6 运行一次 hello.hello_func 方法
scheduler.add_job(hello.hello_func, 'date', run_date=date(2021, 9, 6))
scheduler.add_job(job.hello, 'date', run_date=date(2021, 9, 6), args=['小哈'])

# 在 2021-9-6 09:55:00 运行一次 hello.read_hello 方法
scheduler.add_job(hello.read_hello, 'date', run_date=datetime(2021, 9, 6, 9, 55))
scheduler.add_job(job.hello, 'date', run_date=datetime(2021, 9, 6, 9, 55), args= ['科比',42])

# 在 2021-9-6 09:55:10 运行一次 myapp.myapp_func 方法
scheduler.add_job(myapp.myapp_func, 'date', run_date='2021-9-6 09:55:10')
scheduler.add_job(job.hello, 'date', run_date='2021-9-6 09:55:10', args= ['蓝精灵', 1000, 'unknown'])

scheduler.start()

2、触发器interval

固定时间触发。参数如下：

参数	说明
weeks(int)	间隔几周
days(int)	间隔几天
hours(int)	间隔几小时
minutes(int)	间隔几分钟
seconds(int)	间隔几秒钟
start_date (datetime 或 str)	开始日期
end_date (datetime 或 str)	结束日期
timezone (datetime.tzinfo 或str)	时区

使用例子：

from apscheduler.schedulers.blocking import BlockingScheduler  # 定时任务
from JobStores import job

'''
三种模式:
（1）date: 特定的时间点触发,最基本的一种调度，作业只会执行一次。
（2）interval: 固定时间间隔触发(在某个时间范围内间隔多长时间执行一次)
（3）cron: 在特定时间周期性地触发(cron 和Linux crontab格式兼容，最为强大)
参数说明:
weeks (int): 间隔几周
days (int): 间隔几天
hours (int): 间隔几小时
minutes (int): 间隔几分钟
seconds (int): 间隔多少秒
start_date (datetime 或 str): 开始日期
end_date (datetime 或 str): 结束日期
timezone (datetime.tzinfo 或str): 时区
'''

scheduler = BlockingScheduler()

# 间隔30秒执行一次job.hello
scheduler.add_job(job.hello, 'interval', seconds=30, args=['小哈'])

# 间隔一分钟执行一次job.hello
scheduler.add_job(job.hello, 'interval', minutes=1, args=['科比', 42])

# 间隔一分半执行一次job.hello
scheduler.add_job(job.hello, 'interval', minutes=1, seconds=30, args=['蓝精灵', 1000, 'unknown'])

# 间隔一分半 并且只在[2021-09-06 10:30:00,2021-09-06 10:33:00）]这个时间区间内执行job.hello
scheduler.add_job(job.hello, 'interval', minutes=1, seconds=30, start_date='2021-09-06 10:30:00',
                  end_date='2021-09-06 10:33:00', args=['王铁锤', 38, 'Woman'])

scheduler.start()

# 大家好,我的名字是王铁锤,今年38岁,性别Woman,现在时间为:2021-09-06 10:30:00
# 大家好,我的名字是小哈,今年18岁,性别Man,现在时间为:2021-09-06 10:30:08
# 大家好,我的名字是小哈,今年18岁,性别Man,现在时间为:2021-09-06 10:30:38
# 大家好,我的名字是科比,今年42岁,性别Man,现在时间为:2021-09-06 10:30:38
# 大家好,我的名字是小哈,今年18岁,性别Man,现在时间为:2021-09-06 10:31:08
# 大家好,我的名字是蓝精灵,今年1000岁,性别unknown,现在时间为:2021-09-06 10:31:08
# 大家好,我的名字是王铁锤,今年38岁,性别Woman,现在时间为:2021-09-06 10:31:30
# 大家好,我的名字是小哈,今年18岁,性别Man,现在时间为:2021-09-06 10:31:38
# 大家好,我的名字是科比,今年42岁,性别Man,现在时间为:2021-09-06 10:31:38
# 大家好,我的名字是小哈,今年18岁,性别Man,现在时间为:2021-09-06 10:32:08
# 大家好,我的名字是小哈,今年18岁,性别Man,现在时间为:2021-09-06 10:32:38
# 大家好,我的名字是科比,今年42岁,性别Man,现在时间为:2021-09-06 10:32:38
# 大家好,我的名字是蓝精灵,今年1000岁,性别unknown,现在时间为:2021-09-06 10:32:38
# 大家好,我的名字是王铁锤,今年38岁,性别Woman,现在时间为:2021-09-06 10:33:00

3、触发器cron

在特定时间周期性触发。参数如下：

参数	说明
year (int 或 str)	年，4位数字
month (int 或 str)	月 (范围1-12)
day (int 或 str)	日 (范围1-31)
week (int 或 str)	周 (范围1-53)
day_of_week (int 或 str)	周内第几天或者星期几 (范围0-6 或者 mon,tue,wed,thu,fri,sat,sun)
hour (int 或 str)	时 (范围0-23)
minute (int 或 str)	分 (范围0-59)
second (int 或 str)	秒 (范围0-59)
start_date (datetime 或 str)	最早开始日期(包含)
end_date (datetime 或 str)	最晚结束时间(包含)
timezone (datetime.tzinfo 或str)	指定时区

这些参数支持算数表达式，取值格式有如下：

图片.jpg

使用例子：

from apscheduler.schedulers.blocking import BlockingScheduler  # 定时任务
from JobStores import job

'''
三种模式:
（1）date: 特定的时间点触发,最基本的一种调度，作业只会执行一次。
（2）interval: 固定时间间隔触发(在某个时间范围内间隔多长时间执行一次)
（3）cron: 在特定时间周期性地触发(cron 和Linux crontab格式兼容，最为强大)
参数说明:
year (int 或 str): 年，4位数字
month (int 或 str): 月 (范围1-12)
day (int 或 str): 日 (范围1-31)
week (int 或 str): 周 (范围1-53)
day_of_week (int 或 str): 周内第几天或者星期几 (范围0-6 或者 mon(星期天),tue(星期一),wed(星期二),thu(星期三),fri(星期四),sat(星期五),sun(星期六))
hour (int 或 str): 时 (范围0-23)
minute (int 或 str): 分 (范围0-59)
second (int 或 str): 秒 (范围0-59)
start_date (datetime 或 str): 开始日期
end_date (datetime 或 str): 结束日期
timezone (datetime.tzinfo 或str): 时区
'''

scheduler = BlockingScheduler()

# xx:xx:30执行一次
# scheduler.add_job(job.hello, 'cron', second=30, args=['小哈'])
# 每30秒执行一次(计时起点取整)
# scheduler.add_job(job.hello, 'cron', second='*/30', args=['小呵'])

# 每分钟执行一次
scheduler.add_job(job.hello, 'cron', minute='*/1', args=['科比', 42])
# 在每天11和12点,每分钟执行一次
scheduler.add_job(job.hello, 'cron', hour='11-12', minute='*/1', args=['科比-布莱恩特', 42])
# 在每天11和12点,每10秒执行一次
scheduler.add_job(job.hello, 'cron', hour='11-12', second='*/10', args=['超级赛亚人', 928])
# 在每天11和12点的55和56分,每10秒执行一次
scheduler.add_job(job.hello, 'cron', hour='11-12', minute='55-56', second='*/10', args=['孙悟饭', 200])
# 在每周一的11和12点的55和56分,每10秒执行一次
scheduler.add_job(job.hello, 'cron', day_of_week='1', hour='11-12', minute='55-56', second='*/10',
                  args=['擎天柱', 129, 'unknown'])
# 在每周的周一到周五的11和12点的55和56分,每10秒执行一次
scheduler.add_job(job.hello, 'cron', day_of_week='1-5', hour='11-12', minute='55-56', second='*/10',
                  args=['威震天', 190, 'unknown'])
# 在每月的1号，11号，21号，31号的11和12点的55和56分,每10秒执行一次
scheduler.add_job(job.hello, 'cron', day='*/10', hour='11-12', minute='55-56', second='*/10',
                  args=['大黄蜂', 36, 'unknown'])
# 在每月的1-10号的周一到周五的11和12点的55和56分,每10秒执行一次
scheduler.add_job(job.hello, 'cron', day='1-10', day_of_week='1-5', hour='11-12', minute='55-56', second='*/10',
                  args=['霸天虎', 58, 'unknown'])

scheduler.start()

4、通过装饰器scheduled_job()添加方法

添加任务的方法有两种：

（1）通过调用add_job()---见上面1至3代码

（2）通过装饰器scheduled_job()：第一种方法是最常用的方法。第二种方法主要是方便地声明在应用程序运行时不会更改的任务。该 add_job()方法返回一个apscheduler.job.Job实例，可以使用该实例稍后修改或删除该任务。

使用例子：

import time
from apscheduler.schedulers.blocking import BlockingScheduler

scheduler = BlockingScheduler()

# 每五秒执行一次
@scheduler.scheduled_job('interval', seconds=5)
def job_one():
    t = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
    print(f'job_one --- {t}')

# 每七秒执行一次
@scheduler.scheduled_job('cron', second='*/7')
def job_two():
    t = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
    print(f'job_two --- {t}')


scheduler.start()

# job_two --- 2021-09-07 09:17:07
# job_one --- 2021-09-07 09:17:11
# job_two --- 2021-09-07 09:17:14
# job_one --- 2021-09-07 09:17:16

六、写在最后

入门使用就这些，更高深的用法在实践中去发现吧。。。

THE END

代码编程

二维码

打赏

Python装饰器用法

< <上一篇

Docker之Docker简介

下一篇>>

搜索内容

Python使用APScheduler实现定时任务

一、简介

二、安装APScheduler

三、基本概念

1、触发器 triggers

2、任务储存器 job stores

3、执行器 executors

4、调度器 schedulers

四、使用步骤

五、使用实例

1、触发器date

2、触发器interval

3、触发器cron

4、通过装饰器scheduled_job()添加方法

六、写在最后

取消回复

共有 2 条评论

作者信息

近期文章

剧毒鸡汤

最新评论