Python定时库Apscheduler的简单使用
在Python中需要执行定时任务，可以使用Apscheduler。
Apscheduler是基于Quartz的Python定时任务框架，功能上跟Quartz一致，使用上跟Quartz也几乎一致。
核心的四个部分：
①触发器（trigger）、②作业存储（job store）、③执行器（executor）、④调度器（scheduler）
安装依赖：
pip install apscheduler
间隔时间调度：
from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
sched = BlockingScheduler()
def test_job():
print(f'{datetime.now():%H:%M:%S} Test job')
if __name__ == '__main__':
sched.add_job(test_job, 'interval', id='test', seconds=5)
sched.start()
也可以使用装饰器：
from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
sched = BlockingScheduler()
def test_job():
print(f'{datetime.now():%H:%M:%S} Test job')
@sched.scheduled_job('interval', seconds=5)
def test_decorator_job():
print(f'{datetime.now():%H:%M:%S} Test decorator job')
if __name__ == '__main__':
sched.add_job(test_job, 'interval', id='test', seconds=5)
sched.start()
运行结果：
BlockingScheduler()是调度器中的一种调度器
sched.add_job()是添加作业
sched.start()是开始任务
定时调度：
定时调度使用cron表达式进行，这里也带参数执行：
from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
scheduler = BlockingScheduler()
def test_args(x):
print (f'{datetime.now():%H:%M:%S} Test cron job', x)
if __name__ == '__main__':
scheduler.add_job(test_args, 'cron', args=('cron test',), second='*/5')
scheduler.start()
时间参数设置如下：
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) – 指定时区
接下来说一下其中的调度器：
BlockingScheduler：适用于调度程序是进程中唯一运行的进程，调用start函数会阻塞当前线程，不能立即返回。 BackgroundScheduler：适用于调度程序在应用程序的后台运行，调用start后主线程不会阻塞。 AsyncIOScheduler：适用于使用了asyncio模块的应用程序。 GeventScheduler：适用于使用gevent模块的应用程序。 TwistedScheduler：适用于构建Twisted的应用程序。 QtScheduler：适用于构建Qt的应用程序。
其中使用得比较多的是前三种调度器。
删除任务：
scheduler.remove_job('task_id')
停止任务：
scheduler.pause_job('task_id')
恢复任务：
scheduler.resume_job('task_id')
立即执行任务next_run_time：
scheduler.add_job( test_job, 'interval', minutes=5, next_run_time=datetime.datetime.now() )
到此这篇关于Python定时库Apscheduler的简单使用的文章就介绍到这了,更多相关Python定时库Apscheduler使用内容请搜索本站以前的文章或继续浏览下面的相关文章希望大家以后多多支持本站！
版权声明：本站文章来源标注为YINGSOO的内容版权均为本站所有，欢迎引用、转载，请保持原文完整并注明来源及原文链接。禁止复制或仿造本网站，禁止在非www.yingsoo.com所属的服务器上建立镜像，否则将依法追究法律责任。本站部分内容来源于网友推荐、互联网收集整理而来，仅供学习参考，不代表本站立场，如有内容涉嫌侵权，请联系alex-e#qq.com处理。
关注官方微信