Python任务调度利器：APScheduler

任务调度应用场景

创新互联公司从2013年创立，先为天宁等服务建站，天宁等地企业，进行企业商务咨询服务。为天宁企业网站制作PC+手机+微官网三网同步一站式服务解决您的所有建站问题。

所谓的任务调度是指安排任务的执行计划，即何时执行，怎么执行等。在现实项目中经常出现它们的身影；特别是数据类项目，比如实时统计每5分钟网站的访问量，就需要每5分钟定时从日志数据分析访问量。

总结下任务调度应用场景：

  •  离线作业调度：按时间粒度执行某项任务
  •  共享缓存更新：定时刷新缓存，如redis缓存；不同进程间的共享数据

任务调度工具

  •  linux的crontab， 支持按照分钟/小时/天/月/周粒度，执行任务
  •  java的Quartz
  •  windows的任务计划

本文介绍的是python中的任务调度库，APScheduler（advance python scheduler）。如果你了解Quartz的话，可以看出APScheduler是Quartz的python实现；APScheduler提供了基于时间，固定时间点和crontab方式的任务调用方案， 可以当作一个跨平台的调度工具来使用。

APScheduler

组件介绍

APScheduler由5个部分组成：触发器、调度器、任务存储器、执行器和任务事件。

  •  任务job：任务id和任务执行func
  •  触发器triggers：确定任务何时开始执行
  •  任务存储器job stores: 保存任务的状态
  •  执行器executors：确定任务怎么执行
  •  任务事件event：监控任务执行异常情况
  •  调度器schedulers：串联任务的整个生命周期，添加编辑任务到任务存储器,在任务的执行时间到来时，把任务交给执行器执行返回结果；同时发出事件监听，监控任务事件 。

安装

 
 
 
 
    
  
  
  
  
  1. pip install apscheduler 
    2.

简单例子

 
 
 
 
    
  
  
  
  
  1. from apscheduler.schedulers.background import BackgroundScheduler  
    2.   
  
  
  
  2. from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor  
    3.   
  
  
  
  3. from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore  
    4.   
  
  
  
  4. from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR  
    5.   
  
  
  
  5. import logging  
    6.   
  
  
  
  6. import datetime  
    7.   
  
  
  
  7. # 任务执行函数  
    8.   
  
  
  
  8. def job_func(job_id):  
    9.   
  
  
  
  9.     print('job %s is runed at %s' % (job_id, datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')))  
    10.   
  
  
  
  10. # 事件监听  
    11.   
  
  
  
  11. def job_exception_listener(event):  
    12.   
  
  
  
  12.     if event.exception:  
    13.   
  
  
  
  13.         # todo：异常处理, 告警等  
    14.   
  
  
  
  14.         print('The job crashed :(')  
    15.   
  
  
  
  15.     else:  
    16.   
  
  
  
  16.         print('The job worked :)')  
    17.   
  
  
  
  17. # 日志  
    18.   
  
  
  
  18. logging.basicConfig()  
    19.   
  
  
  
  19. logging.getLogger('apscheduler').setLevel(logging.DEBUG)  
    20.   
  
  
  
  20. # 定义一个后台任务非阻塞调度器  
    21.   
  
  
  
  21. scheduler = BackgroundScheduler()  
    22.   
  
  
  
  22. # 添加一个任务到内存中   
    23.   
  
  
  
  23. # 触发器：trigger='interval' seconds=10 每10s触发执行一次  
    24.   
  
  
  
  24. # 执行器：executor='default' 线程执行  
    25.   
  
  
  
  25. # 任务存储器：jobstore='default' 默认内存存储  
    26.   
  
  
  
  26. # 最大并发数：max_instances  
    27.   
  
  
  
  27. scheduler.add_job(job_func, trigger='interval', args=[1], id='1', name='a test job', max_instances=10, jobstore='default', executor='default', seconds=10)  
    28.   
  
  
  
  28. # 设置任务监听  
    29.   
  
  
  
  29. scheduler.add_listener(job_exception_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)  
    30.   
  
  
  
  30. # 启动调度器  
    31.   
  
  
  
  31. scheduler.start() 
    32.

运行情况：

 
 
 
 
    
  
  
  
  
  1. job 1 is runed at 2020-03-21 20:00:38  
    2.   
  
  
  
  2. The job worked :)  
    3.   
  
  
  
  3. job 1 is runed at 2020-03-21 20:00:48  
    4.   
  
  
  
  4. The job worked :)  
    5.   
  
  
  
  5. job 1 is runed at 2020-03-21 20:00:58  
    6.   
  
  
  
  6. The job worked :) 
    7.

触发器

触发器决定何时执行任务，APScheduler支持的触发器有3种

  •  trigger='interval'：按固定时间周期执行，支持weeks，days，hours，minutes， seconds， 还可指定时间范围   
 
 
 
 
    
  
  
  
  
  1. sched.add_job(job_function, 'interval', hours=2, start_date='2010-10-10 09:30:00', end_date='2014-06-15 11:00:00') 
    2.
  •  trigger='date': 固定时间，执行一次   
 
 
 
 
    
  
  
  
  
  1. sched.add_job(my_job, 'date', run_date=datetime(2009, 11, 6, 16, 30, 5), args=['text']) 
    2.
  •   trigger='cron': 支持crontab方式，执行任务
  •   参数：分钟/小时/天/月/周粒度，也可指定时间范围     
 
 
 
 
    
  
  
  
  
  1. year (int|str) – 4-digit year  
    2.   
  
  
  
  2.       month (int|str) – month (1-12)  
    3.   
  
  
  
  3.       day (int|str) – day of the (1-31)  
    4.   
  
  
  
  4.       week (int|str) – ISO week (1-53)  
    5.   
  
  
  
  5.       day_of_week (int|str) – number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun)  
    6.   
  
  
  
  6.       hour (int|str) – hour (0-23)  
    7.   
  
  
  
  7.       minute (int|str) – minute (0-59)  
    8.   
  
  
  
  8.       second (int|str) – second (0-59)  
    9.   
  
  
  
  9.       start_date (datetime|str) – earliest possible date/time to trigger on (inclusive)  
    10.   
  
  
  
  10.       end_date (datetime|str) – latest possible date/time to trigger on (inclusive) 
    11.
  •   例子           
 
 
 
 
    
  
  
  
  
  1. # 星期一到星期五，5点30执行任务job_function，直到2014-05-30 00:00:00  
    2.   
  
  
  
  2.            sched.add_job(job_function, 'cron', day_of_week='mon-fri', hour=5, minute=30, end_date='2014-05-30')  
    3.   
  
  
  
  3.            # 按照crontab格式执行， 格式为：分钟 小时 天 月 周，*表示所有  
    4.   
  
  
  
  4.            # 5月到8月的1号到15号，0点0分执行任务job_function  
    5.   
  
  
  
  5.            sched.add_job(job_function, CronTrigger.from_crontab('0 0 1-15 may-aug *')) 
    6.

执行器

执行器决定如何执行任务；APScheduler支持4种不同执行器，常用的有pool(线程/进程)和gevent(io多路复用，支持高并发)，默认为pool中线程池， 不同的执行器可以在调度器的配置中进行配置（见调度器）

  •  apscheduler.executors.asyncio：同步io，阻塞
  •  apscheduler.executors.gevent：io多路复用，非阻塞
  •  apscheduler.executors.pool: 线程ThreadPoolExecutor和进程ProcessPoolExecutor
  •  apscheduler.executors.twisted：基于事件驱动

任务存储器

任务存储器决定任务的保存方式， 默认存储在内存中（MemoryJobStore），重启后就没有了。APScheduler支持的任务存储器有：

  •  apscheduler.jobstores.memory：内存
  •  apscheduler.jobstores.mongodb：存储在mongodb
  •  apscheduler.jobstores.redis：存储在redis
  •  apscheduler.jobstores.rethinkdb：存储在rethinkdb
  •  apscheduler.jobstores.sqlalchemy：支持sqlalchemy的数据库如mysql，sqlite等
  •  apscheduler.jobstores.zookeeper：zookeeper

不同的任务存储器可以在调度器的配置中进行配置（见调度器）

调度器

APScheduler支持的调度器方式如下，比较常用的为BlockingScheduler和BackgroundScheduler

  •  BlockingScheduler：适用于调度程序是进程中唯一运行的进程，调用start函数会阻塞当前线程，不能立即返回。
  •  BackgroundScheduler：适用于调度程序在应用程序的后台运行，调用start后主线程不会阻塞。
  •  AsyncIOScheduler：适用于使用了asyncio模块的应用程序。
  •  GeventScheduler：适用于使用gevent模块的应用程序。
  •  TwistedScheduler：适用于构建Twisted的应用程序。
  •  QtScheduler：适用于构建Qt的应用程序。

从前面的例子，我们可以看到，调度器可以操作任务（并为任务指定触发器、任务存储器和执行器）和监控任务。

 
 
 
 
    
  
  
  
  
  1. scheduler.add_job(job_func, trigger='interval', args=[1], id='1', name='a test job', max_instances=10, jobstore='default', executor='default', seconds=10) 
    2.

我们来详细看下各个部分

  •  调度器配置：在add_job我们看到jobstore和executor都是default，APScheduler在定义调度器时可以指定不同的任务存储和执行器，以及初始的参数   
 
 
 
 
    
  
  
  
  
  1. from pytz import utc  
    2.   
  
  
  
  2.    from apscheduler.schedulers.background import BackgroundScheduler  
    3.   
  
  
  
  3.    from apscheduler.jobstores.mongodb import MongoDBJobStore  
    4.   
  
  
  
  4.    from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore  
    5.   
  
  
  
  5.    from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor  
    6.   
  
  
  
  6.    # 通过dict方式执行不同的jobstores、executors和默认的参数  
    7.   
  
  
  
  7.    jobstores = {  
    8.   
  
  
  
  8.        'mongo': MongoDBJobStore(),  
    9.   
  
  
  
  9.        'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')  
    10.   
  
  
  
  10.    }  
    11.   
  
  
  
  11.    executors = {  
    12.   
  
  
  
  12.        'default': ThreadPoolExecutor(20),  
    13.   
  
  
  
  13.        'processpool': ProcessPoolExecutor(5)  
    14.   
  
  
  
  14.    }  
    15.   
  
  
  
  15.    job_defaults = {  
    16.   
  
  
  
  16.        'coalesce': False,  
    17.   
  
  
  
  17.        'max_instances': 3  
    18.   
  
  
  
  18.    }  
    19.   
  
  
  
  19.    # 定义调度器  
    20.   
  
  
  
  20.    scheduler = BackgroundScheduler(jobstoresjobstores=jobstores, executorsexecutors=executors, job_defaultsjob_defaults=job_defaults, timezone=utc)  
    21.   
  
  
  
  21.    def job_func(job_id):  
    22.   
  
  
  
  22.        print('job %s is runed at %s' % (job_id, datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')))  
    23.   
  
  
  
  23.    # 添加任务  
    24.   
  
  
  
  24.    scheduler.add_job(job_func, trigger='interval', args=[1], id='1', name='a test job', jobstore='default', executor='processpool', seconds=10)  
    25.   
  
  
  
  25.    # 启动调度器  
    26.   
  
  
  
  26.    scheduler.start() 
    27.
  •  操作任务：调度器可以增加，删除，暂停，恢复和修改任务。需要注意的是这里的操作只是对未执行的任务起作用，已经执行和正在执行的任务不受这些操作的影响。
  •   add_job     
 
 
 
 
    
  
  
  
  
  1. scheduler.add_job(job_func, trigger='interval', args=[1], id='1', name='a test job', max_instances=10, jobstore='default', executor='default', seconds=10) 
    2.
  •   remove_job: 通过任务唯一的id，删除的时候对应的任务存储器里记录也会删除 
 
 
 
 
    
  
  
  
  
  1. scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id')  
    2.   
  
  
  
  2.  scheduler.remove_job('my_job_id') 
    3.
  •   Pausing and resuming jobs：暂停和重启任务       
 
 
 
 
    
  
  
  
  
  1. scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id')  
    2.   
  
  
  
  2.         scheduler.pause_job('my_job_id')  
    3.   
  
  
  
  3.         scheduler.resume_job('my_job_id') 
    4.
  •   Modifying jobs：修改任务的配置       
 
 
 
 
    
  
  
  
  
  1. job = scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id', max_instances=10)  
    2.   
  
  
  
  2.        # 修改任务的属性  
    3.   
  
  
  
  3.        job.modify(max_instances=6, name='Alternate name')  
    4.   
  
  
  
  4.        # 修改任务的触发器  
    5.   
  
  
  
  5.        scheduler.reschedule_job('my_job_id', trigger='cron', minute='*/5') 
    6.
  •  监控任务事件类型，比较常用的类型有：
    •   EVENT_JOB_ERROR: 表示任务在执行过程的出现异常触发
    •   EVENT_JOB_EXECUTED：任务执行成功时
    •   EVENT_JOB_MAX_INSTANCES：调度器上执行的任务超过配置的参数时       
 
 
 
 
    
  
  
  
  
  1. scheduler.add_listener(job_exception_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)    
    2.

网页名称：Python任务调度利器：APScheduler
文章链接：http://www.shufengxianlan.com/qtweb/news1/182301.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

解决方案知识

行业网站建设

成都微信开发　　 　贵阳联通机房　　 　服务器托管　　 　工业净化空调　　 　LOGO设计　　 　成都门窗定制　　 　美美搬家　　 　重庆水土双线托管　　 　胜锦安防科技　　 　LED显示屏　　 　400电话办理　　 　注册域名　　 　卡多尼人工智　　 　成都棕树电信机房　　 　商城网站　　 　成都网站改版　　 　成都广告制作　　 　雅安网站建设　　 　简阳机房托管　　 　成都企业网站建设