python apscheduler定时任务处理

Apscheduler

Apscheduler是一个基于Quartz的python定时任务框架，提供了基于日期、固定时间间隔以及corntab类型的任务，并且可持久化任务。

安装apscheduler依赖库

pip install apscheduler

APScheduler组件：

• schedulers -调度器
• triggers -触发器
• job stores -作业存储器
• executors -执行器

schedulers（调度器）

• BlockingScheduler 阻塞式调度器：适用于只跑调度器的程序。(常用)
• BackgroundScheduler 后台调度器：适用于非阻塞的情况，调度器会在后台独立运行。(常用)
• AsyncIOScheduler AsyncIO调度器，适用于应用使用AsnycIO的情况。
• GeventScheduler Gevent调度器，适用于应用通过Gevent的情况。
• TornadoScheduler Tornado调度器，适用于构建Tornado应用。
• TwistedScheduler Twisted调度器，适用于构建Twisted应用。
• QtScheduler Qt调度器，适用于构建Qt应用。

triggers（触发器）

1、date 触发器

2、interval 触发器，固定时间间隔触发。

3、cron 触发器，在特定时间周期性地触发，和Linux crontab格式兼容。

job store （作业存储）

该组件是对调度任务的管理，添加 job 有两种添加方法，其中一种是 .add_job()， 另一种则是@scheduled_job()装饰器来修饰函数。

示例1：

# .add_job()

from apscheduler.schedulers.background import BackgroundScheduler
import time


# 定义一个job方法
def job_fun()：
  print("this is a work function job")


if __name__ == '__main__':

    # 创建后台调度器
    scheduler = BlockingScheduler()
    # 添加定时任务，每隔5秒执行一次
    scheduler.add_job(job_fun, 'interval', seconds=5, args=[1], max_instances=3)

    # 启动调度器
    scheduler.start()

示例2：

import time
from apscheduler.schedulers.blocking import BlockingScheduler

scheduler = BlockingScheduler()

# 每隔 5秒钟
@scheduler.scheduled_job('interval', seconds=5)
def job1():    
    print('----- this is interval scheduled_job. -----')

# 每隔 10秒钟
@scheduler.scheduled_job('cron', second='*/10')
def job2():
    print('==== this is cron scheduled_job ====')

scheduler.start()

阻塞和非阻塞

• APScheduler提供了阻塞和非阻塞两种调度器类型，BackgroundScheduler和BlockingScheduler，是两种类型 常用 的调度器，用于创建定时任务。
• 主要区别：BackgroundScheduler在后台运行，不会阻塞主线程；而BlockingScheduler会阻塞主线程直到所有任务完成。

BackgroundScheduler例子

from apscheduler.schedulers.background import BackgroundScheduler
import time


def job():
    print('job 3s')

if __name__=='__main__':

    # BackgroundScheduler调用start函数后并不会阻塞当前线程，所以可以继续执行主程序中while循环的逻辑。
    sched = BackgroundScheduler(timezone='MST')
    sched.add_job(job, 'interval', id='3_second_job', seconds=3)
    sched.start()

    while(True):  # while会被执行
        print('main 1s')
        time.sleep(1)

BlockingScheduler例子：

from apscheduler.schedulers.blocking import BlockingScheduler
import time

def job():
    print('job 3s')

if __name__=='__main__':

    sched = BlockingScheduler(timezone='MST')
    sched.add_job(job, 'interval', id='3_second_job', seconds=3)
    sched.start()

    while(True): # 不会被执行到
        print('main 1s')
        time.sleep(1)