Vue学习之从入门到神经(万字长文 建议收藏)

webpack4中hash、chunkhash、contenthash区别!

  返回  

Python定时库:APScheduler

2021/7/20 15:18:57 浏览:

参考:

  • https://zhuanlan.zhihu.com/p/95563033
  • https://www.cnblogs.com/qq834761298/p/12164717.html
  • https://www.cnblogs.com/KdeS/p/13158152.html#autoid-0-6-0

文章目录

  • 1 基本概念
    • 1.1 基本组件
      • 1.1.1 触发器 triggers
      • 1.1.2 作业储存器 job stores:
      • 1.1.3 执行器 executors
      • 1.1.4 调度器 schedulers:
  • 2 简单示例
    • 2.1 循环执行例子
      • 2.1.1 每5s执行一次
      • 2.1.2 指定时段每分钟执行一次
      • 2.1.3 每天定点执行
      • 2.1.4 每2小时运行一次,但仅在周六和周日运行
  • 3 示例详解
    • 3.1 triggers 使用详解
      • 3.1.1 date 在指定时间点触发任务
      • 3.1.2 datetime类型(用于精确时间)
    • 3.2 interval 周期触发任务
      • 3.2.1 每两小时触发一次
      • 3.2.2 通过scheduled_job()装饰器实现
      • 3.3.3 使用start_date和end_date限制计划运行的时间段
    • 3.3 cron 类crontab表达式

1 基本概念

1.1 基本组件

1.1.1 触发器 triggers

用于设定触发任务的条件

Trigger绑定到Job,在scheduler调度筛选Job时,根据触发器的规则计算出Job的触发时间,然后与当前时间比较确定此Job是否会被执行,总之就是根据trigger规则计算出下一个执行时间。

Trigger有多种种类,指定时间的DateTrigger,指定间隔时间的IntervalTrigger,像 Linux 的 crontab 一样的CronTrigger。

当进行调度作业时,需要为这个作业选择一个triggers,用来描述这个作业何时被触发,APScheduler有三种内置的触发器类型:

  • “date”: 一次性指定日期;
  • “interval” 在某个时间范围内间隔多长时间执行一次;
  • “cron” 和 “Linux crontab” 格式兼容,最为强大。

1.1.2 作业储存器 job stores:

用于存放任务,把任务存放在内存或数据库中

如果你的应用在每次启动的时候都会重新创建作业,那么使用默认的作业存储器(MemoryJobStore)即可,但是如果你需要在调度器重启或者应用程序奔溃的情况下任然保留作业,你应该根据你的应用环境来选择具体的作业存储器。

例如:使用Mongo或者SQLAlchemy JobStore (用于支持大多数RDBMS)

1.1.3 执行器 executors

用于执行任务,可以设定执行模式为单线程或线程池

对执行器的选择取决于你使用上面哪些框架,大多数情况下,使用默认的ThreadPoolExecutor已经能够满足需求。如果你的应用涉及到CPU密集型操作,你可以考虑使用ProcessPoolExecutor来使用更多的CPU核心。你也可以同时使用两者,将ProcessPoolExecutor作为第二执行器。

1.1.4 调度器 schedulers:

把上方三个组件作为参数,通过创建调度器实例来运行

  • BlockingScheduler : 阻塞式调度器:适用于只跑调度器的程序
  • BackgroundScheduler : 后台调度器:适用于非阻塞的情况,调度器会在后台独立运行
  • AsyncIOScheduler : 当你的程序使用了asyncio(一个异步框架)的时候使用。
  • GeventScheduler : 当你的程序使用了gevent(高性能的Python并发框架)的时候使用。
  • TornadoScheduler : 当你的程序基于Tornado(一个web框架)的时候使用。
  • TwistedScheduler : 当你的程序使用了Twisted(一个异步框架)的时候使用
  • QtScheduler : 如果你的应用是一个Qt应用的时候可以使用。

任务储存器的选择,要看任务是否需要持久化。
如果你运行的任务是无状态的,选择默认任务储存器MemoryJobStore就可以应付。
但是,如果你需要在程序关闭或重启时,保存任务的状态,那么就要选择持久化的任务储存器。
作者推荐使用SQLAlchemyJobStore并搭配PostgreSQL作为后台数据库。
这个方案可以提供强大的数据整合与保护功能。

2 简单示例

2.1 循环执行例子

2.1.1 每5s执行一次

#!/usr/bin/python
# -*- coding: UTF-8 -*-
"""
说明:
    - 每5s执行一次
"""

import datetime
from apscheduler.schedulers.blocking import BlockingScheduler


def job1():
    print("Doing Job", datetime.datetime.now())


scheduler = BlockingScheduler()
scheduler.add_job(job1, "interval", seconds=5, id="job1")
# scheduler.add_job(job1, 'cron', minute="/2", id='job1')	# 每隔一分钟执行一次
# scheduler.add_job(job1, 'interval', hours=1, id='job1')	# 每隔一小时执行一次
scheduler.start()


>>>
Doing Job 2021-07-20 14:40:19.232279
Doing Job 2021-07-20 14:40:24.239411
Doing Job 2021-07-20 14:40:29.237285
Doing Job 2021-07-20 14:40:34.244373
Doing Job 2021-07-20 14:40:39.237203
Doing Job 2021-07-20 14:40:44.245174
Doing Job 2021-07-20 14:40:49.242486

2.1.2 指定时段每分钟执行一次

#!/usr/bin/python
# -*- coding: UTF-8 -*-
"""
说明:
    - 在每小时的50-55,每隔一分钟运行一次
"""

import datetime
from apscheduler.schedulers.blocking import BlockingScheduler

scheduler = BlockingScheduler()


@scheduler.scheduled_job("cron", minute="50-55")
def request_update_status():
    print("Doing Job", datetime.datetime.now())


scheduler.start()

2.1.3 每天定点执行

#!/usr/bin/python
# -*- coding: UTF-8 -*-
"""
说明:
    - 每天凌晨1点30分50秒执行一次
"""

import datetime
from apscheduler.schedulers.blocking import BlockingScheduler

scheduler = BlockingScheduler()

f = open('./task_of_scheduler.txt', 'w', encoding='utf8')


@scheduler.scheduled_job("cron", day_of_week="*", hour="1", minute="30", second="30")
def check_db():
    print("Doing Job, and time:", datetime.datetime.now())


if __name__ == '__main__':
    try:
        scheduler.start()
        f.write("定时任务执行成功")
    except Exception as e:
        scheduler.shutdown()
        f.write("定时任务执行失败")
    finally:
        f.close()


2.1.4 每2小时运行一次,但仅在周六和周日运行

#!/usr/bin/python
# -*- coding: UTF-8 -*-
"""
说明:
    - 每2小时运行一次,但仅在周六和周日运行
"""

from apscheduler.triggers.combining import AndTrigger
from apscheduler.triggers.interval import IntervalTrigger
from apscheduler.triggers.cron import CronTrigger

trigger = AndTrigger([IntervalTrigger(hours=2),
                      CronTrigger(day_of_week='sat,sun')])
scheduler.add_job(job_function, trigger)

3 示例详解

3.1 triggers 使用详解

3.1.1 date 在指定时间点触发任务

#!/usr/bin/python
# -*- coding: UTF-8 -*-
"""
说明:
    - # 在2021年7月21日执行
"""

from datetime import date
from apscheduler.schedulers.blocking import BlockingScheduler

scheduler= BlockingScheduler()

def my_job(text):
    print(text)

# 在2021年7月21日执行
scheduler.add_job(my_job, 'date', run_date=date(2021, 7, 21), args=['text'])

scheduler.start()

3.1.2 datetime类型(用于精确时间)

#!/usr/bin/python
# -*- coding: UTF-8 -*-
"""
说明:
    - # 在2021年7月21日执行
"""

from datetime import date, datetime
from apscheduler.schedulers.blocking import BlockingScheduler

scheduler= BlockingScheduler()


def my_job(text):
    print(text, datetime.now())


scheduler.add_job(my_job, 'date', run_date=datetime(2021, 7, 20, 14, 56, 00), args=['Doing Job'])
# scheduler.add_job(my_job, 'date', run_date="2021-07-20 14:56:00", args=['Doing Job'])

scheduler.start()

>>>
Doing Job 2021-07-20 14:56:00.013597

3.2 interval 周期触发任务

3.2.1 每两小时触发一次

#!/usr/bin/python
# -*- coding: UTF-8 -*-
"""
说明:
    - 每两小时触发一次
"""

from datetime import date, datetime
from apscheduler.schedulers.blocking import BlockingScheduler


def job_function():
    print("Hello World", datetime.now())


schedeler = BlockingScheduler()

# 每2小时触发
schedeler.add_job(job_function, 'interval', hours=2)
schedeler.start()

3.2.2 通过scheduled_job()装饰器实现

#!/usr/bin/python
# -*- coding: UTF-8 -*-
"""
说明:
    - 通过scheduled_job()装饰器实现
"""

import time
from apscheduler.schedulers.blocking import BlockingScheduler

scheduler = BlockingScheduler()


@scheduler.scheduled_job('interval', id='my_job_id', minutes=0.5)
def job_function():
    print("Hello World", time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))


scheduler.start()

>>>
Hello World 2021-07-20 15:06:29
Hello World 2021-07-20 15:06:59

3.3.3 使用start_date和end_date限制计划运行的时间段

#!/usr/bin/python
# -*- coding: UTF-8 -*-
"""
说明:
    - 使用start_date和end_date限制计划运行的时间段
"""

import time
from apscheduler.schedulers.blocking import BlockingScheduler

scheduler = BlockingScheduler()


def job_function():
    print("Hello World: ", time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))


scheduler.add_job(job_function, "interval", minutes=0.5,
                  start_date="2021-07-20 15:10:00", 
                  end_date="2021-07-20 15:15:00")
scheduler.start()

>>>
Hello World:  2021-07-20 15:10:00
Hello World:  2021-07-20 15:10:30
Hello World:  2021-07-20 15:11:00
Hello World:  2021-07-20 15:11:30
Hello World:  2021-07-20 15:12:00
Hello World:  2021-07-20 15:12:30
Hello World:  2021-07-20 15:13:00
Hello World:  2021-07-20 15:13:30
Hello World:  2021-07-20 15:14:00
Hello World:  2021-07-20 15:14:30
Hello World:  2021-07-20 15:15:00

3.3 cron 类crontab表达式


"""
当省略时间参数时,在显式指定参数之前的参数会被设定为*,
之后的参数会被设定为最小值,week 和day_of_week的最小值为*。

比如:
设定day=1, minute=20等同于设定year='*', month='*', day=1, week='*', 
day_of_week='*', hour='*', minute=20, second=0,
即每个月的第一天,且当分钟到达20时就触发。
"""
# 注意参数顺序
class apscheduler.triggers.cron.CronTrigger(
year=None, 
month=None, 
day=None, 
week=None, 
day_of_week=None, 
hour=None, 
minute=None,
second=None, 
start_date=None, 
end_date=None, 
timezone=None, 
jitter=None)	# 震动参数

联系我们

如果您对我们的服务有兴趣,请及时和我们联系!

服务热线:18288888888
座机:18288888888
传真:
邮箱:888888@qq.com
地址:郑州市文化路红专路93号