转载之https://lz5z.com/Python%E5%AE%9A%E6%97%B6%E4%BB%BB%E5%8A%A1%E7%9A%84%E5%AE%9E%E7%8E%B0%E6%96%B9%E5%BC%8F/
背景
目前所在的项目组需要经常执行一些定时任务，之前都是用 Node.JS 的 cron来实现 schedule job。可是这次需要连接不同的 DB，而且实现的逻辑也有些许不同，于是选择使用 Python 的定时器。
Python 实现定时任务
循环 sleep
这种方式最简单，在循环里面放入要执行的任务，然后 sleep 一段时间再执行
1
2
3
4
5
6
7
8
9
|
from datetime
import datetime
import time
def timer(n):
while
True:
print(datetime.now().strftime(
"%Y-%m-%d %H:%M:%S"))
time.sleep(n)
timer(
5)
|
这个方法的缺点是，只能执行固定间隔时间的任务，如果有定时任务就无法完成，比如早上六点半喊我起床。并且 sleep 是一个阻塞函数，也就是说 sleep 这一段时间，啥都不能做。
threading模块中的Timer
threading 模块中的 Timer 是一个非阻塞函数，比 sleep 稍好一点，不过依然无法喊我起床。
1
2
3
4
5
6
7
8
9
|
from datetime
import datetime
from threading
import Timer
def printTime(inc):
print(datetime.now().strftime(
"%Y-%m-%d %H:%M:%S"))
t = Timer(inc, printTime, (inc,))
t.start()
printTime(
5)
|
Timer 函数第一个参数是时间间隔（单位是秒），第二个参数是要调用的函数名，第三个参数是调用函数的参数(tuple)
使用sched模块
sched 模块是 Python 内置的模块，它是一个调度（延时处理机制），每次想要定时执行某任务都必须写入一个调度。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
import sched
import time
from datetime
import datetime
schedule = sched.scheduler(time.time, time.sleep)
def printTime(inc):
print(datetime.now().strftime(
"%Y-%m-%d %H:%M:%S"))
schedule.enter(inc,
0, printTime, (inc,))
def main(inc=60):
schedule.enter(
0,
0, printTime, (inc,))
schedule.run()
main(
10)
|
sched 使用步骤如下：
（1）生成调度器： s = sched.scheduler(time.time,time.sleep) 第一个参数是一个可以返回时间戳的函数，第二个参数可以在定时未到达之前阻塞。
（2）加入调度事件 其实有 enter、enterabs 等等，我们以 enter 为例子。 s.enter(x1,x2,x3,x4) 四个参数分别为：间隔事件、优先级（用于同时间到达的两个事件同时执行时定序）、被调用触发的函数，给触发函数的参数（注意：一定要以 tuple 给，如果只有一个参数就(xx,)）
（3）运行 s.run() 注意 sched 模块不是循环的，一次调度被执行后就 Over 了，如果想再执行，请再次 enter
APScheduler定时框架
终于找到了可以每天定时喊我起床的方式了
APScheduler是一个 Python 定时任务框架，使用起来十分方便。提供了基于日期、固定时间间隔以及 crontab 类型的任务，并且可以持久化任务、并以 daemon 方式运行应用。
使用 APScheduler 需要安装
1
|
$
pip install apscheduler
|
首先来看一个周一到周五每天早上6点半喊我起床的例子
1
2
3
4
5
6
7
8
9
|
from apscheduler.schedulers.blocking
import BlockingScheduler
from datetime
import datetime
def job():
print(datetime.now().strftime(
"%Y-%m-%d %H:%M:%S"))
scheduler = BlockingScheduler()
scheduler.add_job(job,
'cron', day_of_week=
'1-5', hour=
6, minute=
30)
scheduler.start()
|
代码中的 BlockingScheduler 是什么呢？
BlockingScheduler 是 APScheduler 中的调度器，APScheduler 中有两种常用的调度器，BlockingScheduler 和 BackgroundScheduler，当调度器是应用中唯一要运行的任务时，使用 BlockingSchedule，如果希望调度器在后台执行，使用 BackgroundScheduler。
- BlockingScheduler: use when the scheduler is the only thing running in your process
- BackgroundScheduler: use when 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
APScheduler四个组件
APScheduler 四个组件分别为：触发器(trigger)，作业存储(job store)，执行器(executor)，调度器(scheduler)。
触发器(trigger)
包含调度逻辑，每一个作业有它自己的触发器，用于决定接下来哪一个作业会运行。除了他们自己初始配置意外，触发器完全是无状态的 APScheduler 有三种内建的 trigger:
date: 特定的时间点触发 interval: 固定时间间隔触发 cron: 在特定时间周期性地触发
作业存储(job store)
存储被调度的作业，默认的作业存储是简单地把作业保存在内存中，其他的作业存储是将作业保存在数据库中。一个作业的数据讲在保存在持久化作业存储时被序列化，并在加载时被反序列化。调度器不能分享同一个作业存储。 APScheduler 默认使用 MemoryJobStore，可以修改使用 DB 存储方案
执行器(executor)
处理作业的运行，他们通常通过在作业中提交制定的可调用对象到一个线程或者进城池来进行。当作业完成时，执行器将会通知调度器。 最常用的 executor 有两种：
ProcessPoolExecutor ThreadPoolExecutor
调度器(scheduler)
通常在应用中只有一个调度器，应用的开发者通常不会直接处理作业存储、调度器和触发器，相反，调度器提供了处理这些的合适的接口。配置作业存储和执行器可以在调度器中完成，例如添加、修改和移除作业。
配置调度器
APScheduler提供了许多不同的方式来配置调度器，你可以使用一个配置字典或者作为参数关键字的方式传入。你也可以先创建调度器，再配置和添加作业，这样你可以在不同的环境中得到更大的灵活性。
下面来看一个简单的 BlockingScheduler 例子
1
2
3
4
5
6
7
8
9
|
from apscheduler.schedulers.blocking
import BlockingScheduler
from datetime
import datetime
def job():
print(datetime.now().strftime(
"%Y-%m-%d %H:%M:%S"))
sched = BlockingScheduler()
sched.add_job(job,
'interval', seconds=
5)
sched.start()
|
上述代码创建了一个 BlockingScheduler，并使用默认内存存储和默认执行器。(默认选项分别是 MemoryJobStore 和 ThreadPoolExecutor，其中线程池的最大线程数为10)。配置完成后使用 start() 方法来启动。
如果想要显式设置 job store(使用mongo存储)和 executor 可以这样写：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
|
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
host =
'127.0.0.1'
port =
27017
client = MongoClient(host, port)
def job():
print(datetime.now().strftime(
"%Y-%m-%d %H:%M:%S"))
jobstores = {
'mongo': MongoDBJobStore(collection=
'job', database=
'test', 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(job,
'interval', seconds=
5, jobstore=
'mongo')
scheduler.start()
|
在运行程序5秒后，第一次输出时间。 在 MongoDB 中可以看到 job 的状态
对 job 的操作
添加 job
添加job有两种方式：
- add_job()
- scheduled_job()
第二种方法只适用于应用运行期间不会改变的 job，而第一种方法返回一个apscheduler.job.Job 的实例，可以用来改变或者移除 job。
1
2
3
4
5
6
7
8
|
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() 返回的实例
1
2
3
4
5
|
job = scheduler.add_job(myfunc,
'interval', minutes=
2)
job.remove()
scheduler.add_job(myfunc,
'interval', minutes=
2, id=
'my_job_id')
scheduler.remove_job(
'my_job_id')
|
暂停和恢复 job
暂停一个 job：
1
2
|
apscheduler.job.Job.pause()
apscheduler.schedulers.base.BaseScheduler.pause_job()
|
恢复一个 job：
1
2
|
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 的属性
1
2
|
job.modify(max_instances=
6, name=
'Alternate name')
modify_job(
'my_job_id', trigger=
'cron', minute=
'*/5')
|
关闭 job
默认情况下调度器会等待所有的 job 完成后，关闭所有的调度器和作业存储。将 wait 选项设置为 False 可以立即关闭。
1
2
|
scheduler.shutdown()
scheduler.shutdown(wait=
False)
|
scheduler 事件
scheduler 可以添加事件监听器，并在特殊的时间触发。
1
2
3
4
5
6
7
|
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 规则
最基本的一种调度，作业只会执行一次。它的参数如下：
- 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
1
2
3
4
5
6
7
8
9
10
11
12
|
from datetime
import date
from apscheduler.schedulers.blocking
import BlockingScheduler
sched = BlockingScheduler()
def my_job(text):
print(text)
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'])
版权声明：本文来源CSDN，感谢博主原创文章，遵循 CC 4.0 by-sa 版权协议，转载请附上原文出处链接和本声明。
原文链接：https://blog.csdn.net/gujing001/article/details/76262311
站方申明：本站部分内容来自社区用户分享，若涉及侵权，请联系站方删除。
-
发表于 2021-06-12 18:48:53
- 阅读 ( 908 )
- 分类：
如果觉得我的文章对您有用，请随意打赏。你的支持将鼓励我继续创作！
|
