Python的一个定时任务框架APScheduler

时间:2022-04-24 07:46:56

APScheduler是基于Quartz(java实现的一个开源作业调度框架)的一个Python定时任务框架,实现了Quartz的所有功能,使用起来十分方便

项目地址 https://bitbucket.org/agronholm/apscheduler/

文档 http://apscheduler.readthedocs.org/en/3.0/

 

简单查看文档

APScheduler has four kinds of components:

  • triggers
  • job stores
  • executors
  • schedulers

Here’s a quick guide for choosing a scheduler

  • BlockingScheduler: use when the scheduler is the only thing running in your process
  • BackgroundScheduler: use then you’re not using any of the frameworks below, and want the scheduler to run in the background inside your application
  • AsyncIOScheduler: use if your application uses the asyncio module
  • GeventScheduler: use if your application uses gevent
  • TornadoScheduler: use if you’re building a Tornado application
  • TwistedScheduler: use if you’re building a Twisted application
  • QtScheduler: use if you’re building a Qt application

常见的几种调度栗子

apschedules/examples/

https://bitbucket.org/agronholm/apscheduler/src/a6545cf29831bb6d8f055aeb2fa9d5204e9bd192/examples/?at=master

 

参照栗子,我可以迅速的写出最常规的BlockingScheduler调度

from datetime import datetime
import os

from apscheduler.schedulers.blocking import BlockingScheduler

def tick():
    print('Tick! The time is: %s' % datetime.now())


if __name__ == '__main__':
    scheduler = BlockingScheduler()
    scheduler.add_job(tick, 'interval', seconds=3)
    print('Press Ctrl+{0} to exit'.format('Break' if os.name == 'nt' else 'C'))

    try:
        scheduler.start()
    except (KeyboardInterrupt, SystemExit):
        pass

add_job函数这里有一个max_instances参数用于指定当前工作同时运行的实例数量,默认为1,

但是可能的上一个工作正在运行而未结束,那么下一个就认为失败, 那么此时可以为一个特定的作业设置最大数量的调度程序

如:

scheduler.add_job(tick, 'interval', seconds=3, max_instances=6)
job.modify(max_instances=6, name='Alternate name')


更多的信息参考文档以及examples。