python实现定时任务那些你不知道的模块

1、使用time中的sleep

这种方式最简单，在循环里放入要执行的任务，而后sleep一段时间在执行html

from datetime import datetime import time # 每n秒执行一次
def timer(n): while True: print(datetime.now().strftime("%Y-%m-%d %H:%M:%S")) time.sleep(n) # 5s
timer(5)

这个方法的缺点是：只能执行固定时间间隔的任务，若是有定时任务就没法完成，好比早上六点半喊我起床，而且sleep是一个阻塞函数，也就是在sleep的这段时间只能等待，什么是也作不了python

2、threading模块中的Timer

threading模块中的Timer是一个非阻塞函数，比sleep稍好一点，不过依然没法叫你起床git

from datetime import datetime from threading import Timer # 打印时间函数
def printTime(n): print(datetime.now().strftime("%Y-%m-%d %H:%M:%S")) t = Timer(n, printTime, (n,)) t.start() # 2s
printTime(2)

Timer类第一个参数是时间间隔(单位是秒)，第二个参数是要盗用的函数名，第三个参数调用函数的参数(是一个tuple)mongodb

3、sched模块

sched模块是python内置的模块，它是一个调度(延时处理机制)，每次想要定时执行某任务必需要写一个调度数据库

import sched import time from datetime import datetime # 初始化sched模块的 scheduler 类 # 第一个参数是一个能够返回时间戳的函数，第二个参数能够在定时未到达以前阻塞。
schedule = sched.scheduler(time.time, time.sleep) # 被周期性调度触发的函数
def printTime(n): print(datetime.now().strftime("%Y-%m-%d %H:%M:%S")) schedule.enter(n, 0, printTime, (n,)) # 每隔10秒执行一个printTime函数


# 默认参数5s
def main(n=5): # enter四个参数分别为：间隔时间、优先级（用于同时间到达的两个事件同时执行时定序）、被调用触发的函数，
    # 给该触发函数的参数（tuple形式）
    schedule.enter(6, 0, printTime, (n,))  # 6秒后执行一次printTime函数
 schedule.run() # 10s 输出一次
main(10)

sched 使用步骤以下：app

（1）生成调度器：
s = sched.scheduler(time.time,time.sleep)
第一个参数是一个能够返回时间戳的函数，第二个参数能够在定时未到达以前阻塞。框架

（2）加入调度事件
其实有 enter、enterabs 等等，咱们以 enter 为例子。
s.enter(x1,x2,x3,x4)
四个参数分别为：间隔事件、优先级（用于同时间到达的两个事件同时执行时定序）、被调用触发的函数，给触发函数的参数（注意：必定要以 tuple 给，若是只有一个参数就(xx,)）async

（3）运行
s.run()
注意： sched 模块不是循环的，一次调度被执行后就 Over 了，若是想再执行，请再次 enteride

4、APScheduler定时框架

终于到了你要找的能够叫你起床的定时任务了函数

APScheduler是一个python定时任务框架，使用起来也十分方面，提供了基于日期，固定时间间隔以及crontab类型的任务，而且能够持久化任务，兵役daemon(守护进程)方式运行应用

使用APSchduler模块须要先进性安装

pip3 install apscheduler

下面是实现一个在周一到周五指定的时刻执行任务

from apscheduler.schedulers.blocking import BlockingScheduler from datetime import datetime # 输出时间
def job(): print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))  # 周一到周五的上午十点四十三分叫我起床


# BlockingScheduler
scheduler = BlockingScheduler() scheduler.add_job(job, 'cron', day_of_week='1-5', hour=10, minute=43) scheduler.start()

上述内容进行说明：

　　代码中的 BlockingScheduler 是什么呢？

　　BlockingScheduler 是 APScheduler 中的调度器，APScheduler 中有两种经常使用的调度器，BlockingScheduler 和 BackgroundScheduler，当调度器是应用中惟一要运行的任务时，使用　　BlockingSchedule，若是但愿调度器在后台执行，使用 BackgroundScheduler。

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

上面的列子已经知足咱们的基本使用，若是想深刻了解请看下面内容：

APScheduler四个组件

APScheduler 四个组件分别为：触发器(trigger)，做业存储(job store)，执行器(executor)，调度器(scheduler)。

触发器(trigger)

包含调度逻辑，每个做业有它本身的触发器，用于决定接下来哪个做业会运行。除了他们本身初始配置意外，触发器彻底是无状态的
APScheduler 有三种内建的 trigger:

date: 特定的时间点触发
interval: 固定时间间隔触发
cron: 在特定时间周期性地触发

做业存储(job store)

存储被调度的做业，默认的做业存储是简单地把做业保存在内存中，其余的做业存储是将做业保存在数据库中。一个做业的数据讲在保存在持久化做业存储时被序列化，并在加载时被反序列化。调度器不能分享同一个做业存储。
APScheduler 默认使用 MemoryJobStore，能够修改使用 DB 存储方案

执行器(executor)

处理做业的运行，他们一般经过在做业中提交制定的可调用对象到一个线程或者进城池来进行。看成业完成时，执行器将会通知调度器。
最经常使用的 executor 有两种：

ProcessPoolExecutor
ThreadPoolExecutor

调度器(scheduler)

一般在应用中只有一个调度器，应用的开发者一般不会直接处理做业存储、调度器和触发器，相反，调度器提供了处理这些的合适的接口。配置做业存储和执行器能够在调度器中完成，例如添加、修改和移除做业。

配置调度器

APScheduler提供了许多不一样的方式来配置调度器，你可使用一个配置字典或者做为参数关键字的方式传入。你也能够先建立调度器，再配置和添加做业，这样你能够在不一样的环境中获得更大的灵活性。

执行时间间隔执行一次任务

from apscheduler.schedulers.blocking import BlockingScheduler from datetime import datetime def func(): print(datetime.now().strftime("%Y-%m-%d %H:%M:%S")) # 定义BlockingScheduler
sched = BlockingScheduler() sched.add_job(func, 'interval', seconds=5)  # 每一个5秒执行一次func函数
sched.start()

上述代码建立了一个 BlockingScheduler，并使用默认内存存储和默认执行器。(默认选项分别是 MemoryJobStore 和 ThreadPoolExecutor，其中线程池的最大线程数为10)。配置完成后使用 start() 方法来启动。

若是想要显式设置 job store(使用mongo存储)和 executor 能够这样写：

from datetime import datetime from pymongo import MongoClient from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.jobstores.memory import MemoryJobStore from apscheduler.jobstores.mongodb import MongoDBJobStore from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor # MongoDB 参数
host = '127.0.0.1' port = 27017 client = MongoClient(host, port) # 输出时间
def func(): print(datetime.now().strftime("%Y-%m-%d %H:%M:%S")) # 存储方式
jobstores = { 'mongo': MongoDBJobStore(collection='test', database='local', client=client), 'default': MemoryJobStore() } executors = { 'default': ThreadPoolExecutor(10), 'processpool': ProcessPoolExecutor(3) } job_defaults = { 'coalesce': False, 'max_instances': 3 } scheduler = BlockingScheduler(jobstores=jobstores, executors=executors, job_defaults=job_defaults) scheduler.add_job(func, 'interval', seconds=5, jobstore='mongo')  # 每隔5秒执行一个func函数
scheduler.start()

在运行程序5秒后，第一次输出时间。
在 MongoDB 中能够看到 job 的状态

对 job 的操做

添加 job

添加job有两种方式：

add_job()
scheduled_job()

第二种方法只适用于应用运行期间不会改变的 job，而第一种方法返回一个apscheduler.job.Job 的实例，能够用来改变或者移除 job。

from apscheduler.schedulers.blocking import BlockingScheduler sched = BlockingScheduler() # 装饰器
@sched.scheduled_job('interval', id='my_job_id', seconds=5) def job_function(): print("Hello World") # 开始
sched.start()

@sched.scheduled_job() 是 Python 的装饰器。

移除 job

移除 job 也有两种方法：

remove_job()
job.remove()

remove_job 使用 jobID 移除
job.remove() 使用 add_job() 返回的实例

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

暂停和恢复 job

暂停一个 job：

apscheduler.job.Job.pause() apscheduler.schedulers.base.BaseScheduler.pause_job()

恢复一个 job：

apscheduler.job.Job.resume() apscheduler.schedulers.base.BaseScheduler.resume_job()

但愿你还记得 apscheduler.job.Job 是 add_job() 返回的实例

获取 job 列表

得到可调度 job 列表，可使用get_jobs() 来完成，它会返回全部的 job 实例。

也可使用print_jobs() 来输出全部格式化的 job 列表。

修改 job

除了 jobID 以外 job 的全部属性均可以修改，使用 apscheduler.job.Job.modify() 或者 modify_job() 修改一个 job 的属性

job.modify(max_instances=6, name='Alternate name') modify_job('my_job_id', trigger='cron', minute='*/5')

关闭 job

默认状况下调度器会等待全部的 job 完成后，关闭全部的调度器和做业存储。将 wait 选项设置为 False 能够当即关闭。

scheduler.shutdown() scheduler.shutdown(wait=False)

scheduler 事件

scheduler 能够添加事件监听器，并在特殊的时间触发。

def my_listener(event): if event.exception: print('The job crashed :(') else: print('The job worked :)') # 添加监听器
scheduler.add_listener(my_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)

trigger 规则

date

最基本的一种调度，做业只会执行一次。它的参数以下：

run_date (datetime|str) – the date/time to run the job at
timezone (datetime.tzinfo|str) – time zone for run_date if it doesn’t have one already

from datetime import date from apscheduler.schedulers.blocking import BlockingScheduler sched = BlockingScheduler() def my_job(text): print(text) # The job will be executed on November 6th, 2009
sched.add_job(my_job, 'date', run_date=date(2009, 11, 6), args=['text']) sched.add_job(my_job, 'date', run_date=datetime(2009, 11, 6, 16, 30, 5), args=['text']) sched.add_job(my_job, 'date', run_date='2009-11-06 16:30:05', args=['text']) # The 'date' trigger and datetime.now() as run_date are implicit
sched.add_job(my_job, args=['text']) sched.start()

cron

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)
timezone (datetime.tzinfo|str) – time zone to use for the date/time calculations (defaults to scheduler timezone)

表达式:

from apscheduler.schedulers.blocking import BlockingScheduler def job_function(): print("Hello World") # BlockingScheduler
sched = BlockingScheduler() # Schedules job_function to be run on the third Friday # of June, July, August, November and December at 00:00, 01:00, 02:00 and 03:00
sched.add_job(job_function, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3') # Runs from Monday to Friday at 5:30 (am) until 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') sched.start()

interval

参数：

weeks (int) – number of weeks to wait
days (int) – number of days to wait
hours (int) – number of hours to wait
minutes (int) – number of minutes to wait
seconds (int) – number of seconds to wait
start_date (datetime|str) – starting point for the interval calculation
end_date (datetime|str) – latest possible date/time to trigger on
timezone (datetime.tzinfo|str) – time zone to use for the date/time calculations

from datetime import datetime from apscheduler.schedulers.blocking import BlockingScheduler def job_function(): print("Hello World") # BlockingScheduler
sched = BlockingScheduler() # Schedule job_function to be called every two hours
sched.add_job(job_function, 'interval', hours=2) # The same as before, but starts on 2010-10-10 at 9:30 and stops on 2014-06-15 at 11:00
sched.add_job(job_function, 'interval', hours=2, start_date='2010-10-10 09:30:00', end_date='2014-06-15 11:00:00') sched.start()