Sanic 框架是一个比较新的Python Web服务框架,利用强大的异步实现高速的访问。用官方的话就是“Build fast. Run fast.” ,号称“下一代 Python Web 服务器/框架”。
根据之前一段时间的使用体验,整体性能还不错,便捷性也不错。基于一些使用场景,遇到了需要做一些定时任务的场景,之前也总结了一下做定时任务的内容,内容在这里:
Sanic 结合APScheduler做定时任务处理_sanic 定时器_czwhit的博客-CSDN博客
近期再次查看的时候发现有一些需要优化的内容,因此针对上面文章不完善的地方在此处做一些补充。
为什么特意要做这些补充?结合这几天查找优化方案的情况来看,网上能找到的大部分内容都是类似的,出于Sanic官方的说明,即给出的代码基本是这种:
async def explicit_inject(app):
await asyncio.sleep(5)
print(app.name)
app.add_task(explicit_inject(app))
这种采用间隔时间法来实现定时任务,一般是比较简单的场景可以使用。对于一些稍微复杂的场景采用asyncio.sleep(x)的方式就不太容易实现了,比如:每周一7点、周三12点X分执行任务等场景。
BlockingScheduler
因此还是基于前面使用过的APScheduler来实现;之前的不好的示范,是这么写的:
from apscheduler.schedulers.blocking import BlockingScheduler
async def tick():
logger.info(f'Tick! 当前时间:{time.strftime("%Y%m%d%H%M%S", time.localtime())} ')
async def taskJob(app):
while True:
sched = BlockingScheduler()
# 每十分钟执行一次任务
sched.add_job(tick, 'interval', minutes=15, id='task_job', jitter=120)
logger.info(f"taskjob定时任务一执行成功!")
# 在每天7点,每隔 5分钟 运行一次 job 方法
# sched.add_job(server_state, 'cron', hour=7, minute='*/5')
sched.add_job(tick, 'cron', hour=15, minute='45')
# 在每天16和17点的10分,运行一次 job 方法
sched.add_job(tick, 'cron', hour='16-17', minute='10')
await sched.start()
app.add_task(taskJob(app))
上面代码中调用的是APScheduler 的 BlockingScheduler 函数,可以实现定时功能;但是由于BlockingScheduler是一个非异步的程序,会阻塞Sanic的正常应用,倒是其他地址无法访问。如果只有这一个定时应用的话,没有其他应用,使用此方法也是可行的。
AsyncIOScheduler
结合sanic的异步特色,APScheduler有一个AsyncIOScheduler函数,是异步函数,因此直接选用。
因此经过改进和验证后,比较好的时间方案是可以通过下面的方式实现:
from apscheduler.schedulers.asyncio import AsyncIOScheduler
async def sche(app):
scheduler = AsyncIOScheduler()
# 每十分钟执行一次任务
scheduler.add_job(server_state, 'interval', minutes=10, id='my_job', jitter=120)
logger.info(f'定时任务二,执行完成!')
scheduler.start()
app.add_task(sche(app))
百度搜到的方案基本都使用了 :while True 来实现,这个对于非异步的python脚本应用没问题;当结合Sanic做异步应用时,就不再需要这个了。
经过实践测试,上面的这段代码运行的良好。
Sanic 框架结合APScheduler实现异步定时任务,现在有了一个比较好的实现方案,与大家分享一下。