python中定时任务使用

一、简介

python中定时任务使用apscheduler(advanced python scheduler)。

二、知识点

2.1 文档官网地址

文档官网地址为：https://apscheduler.readthedocs.io/en/latest/userguide.html

2.2 安装

pip install apscheduler

2.3 组合

apscheduler由作业(job)、作业存储(jobstore)、调度器(scheduler)、执行器(executor)、触发器(trigger)、监听器(listener)组成。

2.3.1 作业

作业(job)其实就是自定义的函数。

2.3.2 作业存储

作业存储(jobstore)指作业存储的位置，可以在内存中，也可以持久化到其它存储中，常用如：

MemoryJobStore：默认
MongoDBJobStore
RedisJobStore
RethinkDBJobStore
SQLAlchemyJobStore
ZooKeeperJobStore

2.3.4 调度器

调度器(scheduler)用于作业的添加、修改、删除等操作，常用调度器有：

BlockingScheduler: 当前程序中使用,调用start方法后会阻塞主线程
BackgroundScheduler: 程序后台使用,调用start方法后不会阻塞主线程
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

2.2.5 执行器

执行器(executor)用于作业的真正执行，通常是在线程池中完成。

ThreadPoolExecutor：默认
ProcessPoolExecutor：适用于cpu密集型
AsyncIOExecutor
GeventExecutor
TornadoExecutor
TwistedExecutor

2.3.6 触发器

触发器(trigger)触发作业的执行，即指定作业什么时候执行，常用有：

date:在指定时间执行
interval: 指定间隔时间执行
cron: 根据cron规则执行

2.3.6 监听器

监听器(listener)用于监听事件，也是一个函数，用于在特定事件执行自定义操作。监听事件有：

'EVENT_SCHEDULER_STARTED', 'EVENT_SCHEDULER_SHUTDOWN', 'EVENT_SCHEDULER_PAUSED',
'EVENT_SCHEDULER_RESUMED', 'EVENT_EXECUTOR_ADDED', 'EVENT_EXECUTOR_REMOVED',
'EVENT_JOBSTORE_ADDED', 'EVENT_JOBSTORE_REMOVED', 'EVENT_ALL_JOBS_REMOVED',
'EVENT_JOB_ADDED', 'EVENT_JOB_REMOVED', 'EVENT_JOB_MODIFIED', 'EVENT_JOB_EXECUTED',
'EVENT_JOB_ERROR', 'EVENT_JOB_MISSED', 'EVENT_JOB_SUBMITTED', 'EVENT_JOB_MAX_INSTANCES',
'SchedulerEvent', 'JobEvent', 'JobExecutionEvent', 'JobSubmissionEvent'

2.4 核心函数

BaseScheduler下的add_job为添加作业方法，定义如下：

def add_job(self, func, trigger=None, args=None, kwargs=None, id=None, name=None,misfire_grace_time=undefined, coalesce=undefined, max_instances=undefined,next_run_time=undefined, jobstore='default', executor='default',replace_existing=False, **trigger_args):

核心参数解释如下：

id:作业的唯一id
name:作业名称
func：需执行的job函数
kwargs：需执行的job函数的传入参数
trigger:触发器
executor:执行器
next_run_time：job的执行时间，不指定则默认由trigger获取

2.5 开发步骤

定时任务开发步骤如下：

1. 定义调度器，另外还可指定执行器(默认为ThreadPoolExecutor)和作业存储(默认为MemoryJobStore)，如：

   sc = BlockingScheduler()
   

2. 定义任务，其实也就是一个函数

3. 根据需要定义监听器并添加到调度器中

4. 将任务添加到调度器中，同时指定触发器

三、示例

3.1 常规示例

import time
from datetime import datetime
from apscheduler.schedulers.blocking import BlockingScheduler# 定义调度器
sc = BlockingScheduler()# 定义任务
def task():print(datetime.now())# 以装饰器方式执行定时任务任务，每隔2秒执行一次
@sc.scheduled_job('cron', id='my_job_id', second='*/2')
def annotation_task():print(datetime.now())def task_exe():# 不指定触发器，则立即执行sc.add_job(task, id='id_now')# interval方式,间隔执行，每隔2秒执行一次sc.add_job(task, 'interval', seconds=2, id='id_interval_exe')# interval方式,间隔执行，每隔2秒执行一次,且指定执行时间段sc.add_job(task, 'interval', seconds=2, start_date='2019-07-24 18:34:00', end_date='2019-07-24 18:50:00',id='id_interval_exe_1')# cron方式，根据cron执行sc.add_job(task, 'cron', second=1, id='id_cron_exe')sc.add_job(task, 'cron', second="*/3", id='id_cron_exe')# date方式,在指定时间点执行sc.add_job(task, 'date', run_date=datetime(2019, 7, 24, 18, 18, 18), id='id_cron_exe')sc.add_job(task, 'date', run_date='2019-07-24 18:21:18', id='id_cron_exe_one')# 启动调度器try:sc.start()except (KeyboardInterrupt, SystemExit):sc.shutdown()time.sleep(20)# 移出任务sc.remove_job(id='id_interval_exe')def main():task_exe()if __name__ == '__main__':main()

3.2 任务操作及监听器示例

import time
from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR
from apscheduler.schedulers.background import BackgroundScheduler# 定义任务
def helloTask(name):print("hello {}".format(name))time.sleep(1)# 监听器
def helloListener(event):if event.code == EVENT_JOB_EXECUTED:print(event)def backgroud_interval_exe():# 定义调度器，非阻塞方式backgroundScheduler = BackgroundScheduler()# 添加监听器backgroundScheduler.add_listener(helloListener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)#添加任务backgroundScheduler.add_job(helloTask, 'interval',id='helloTask_1', seconds=1, kwargs={"name":"yes"})#启动调度器backgroundScheduler.start()time.sleep(3)#暂停任务backgroundScheduler.pause_job('helloTask_1')time.sleep(3)#恢复任务backgroundScheduler.resume_job('helloTask_1')time.sleep(3)# 移出任务backgroundScheduler.remove_job('helloTask_1')time.sleep(3)# 关闭调度器backgroundScheduler.shutdown()def main():backgroud_interval_exe()if __name__ == '__main__':main()