Python任务调度利器之APScheduler详解

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

year (int|str) ? 4-digit year 
 month (int|str) ? month (1-12) 
 day (int|str) ? day of the (1-31) 
 week (int|str) ? ISO week (1-53) 
 day_of_week (int|str) ? number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun) 
 hour (int|str) ? hour (0-23) 
 minute (int|str) ? minute (0-59) 
 second (int|str) ? second (0-59) 
 start_date (datetime|str) ? earliest possible date/time to trigger on (inclusive) 
 end_date (datetime|str) ? latest possible date/time to trigger on (inclusive) 

# 星期一到星期五，5点30执行任务job_function，直到2014-05-30 00:00:00 
  sched.add_job(job_function, 'cron', day_of_week='mon-fri', hour=5, minute=30, end_date='2014-05-30') 
  # 按照crontab格式执行， 格式为：分钟 小时 天 月 周，*表示所有 
  # 5月到8月的1号到15号，0点0分执行任务job_function 
  sched.add_job(job_function, CronTrigger.from_crontab('0 0 1-15 may-aug *')) 

from pytz import utc 
 from apscheduler.schedulers.background import BackgroundScheduler 
 from apscheduler.jobstores.mongodb import MongoDBJobStore 
 from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore 
 from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor 
 # 通过dict方式执行不同的jobstores、executors和默认的参数 
 jobstores = { 
 'mongo': MongoDBJobStore(), 
 'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite') 
 } 
 executors = { 
 'default': ThreadPoolExecutor(20), 
 'processpool': ProcessPoolExecutor(5) 
 } 
 job_defaults = { 
 'coalesce': False, 
 'max_instances': 3 
 } 
 # 定义调度器 
 scheduler = BackgroundScheduler(jobstoresjobstores=jobstores, executorsexecutors=executors, job_defaultsjob_defaults=job_defaults, timezone=utc) 
 def job_func(job_id): 
 print('job %s is runed at %s' % (job_id, datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'))) 
 # 添加任务 
 scheduler.add_job(job_func, trigger='interval', args=[1], id='1', name='a test job', jobstore='default', executor='processpool', seconds=10) 
 # 启动调度器 
 scheduler.start() 

scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id') 
 scheduler.remove_job('my_job_id') 

scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id') 
 scheduler.pause_job('my_job_id') 
 scheduler.resume_job('my_job_id') 

job = scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id', max_instances=10) 
 # 修改任务的属性 
 job.modify(max_instances=6, name='Alternate name') 
 # 修改任务的触发器 
 scheduler.reschedule_job('my_job_id', trigger='cron', minute='*/5')