python apscheduler 堵塞_python使用Apscheduler定时清理数据

布满荆棘的人生 2022-10-31 13:43 308阅读 0赞

Apscheduler简介

Apscheduler我就不过多赘述了。如果你还不太了解Apscheduler，请参考如下链接：

应用场景示例

话不多说，如下场景:

需要清理某个数据库的数据，保留半年数据，最好在周末自动执行。

使用Apscheduler则可以完美解决该问题。

使用mongodb管理定时任务

def get\_scheduler():

"""mongodb存储job,返回任务调度器"""

scheduler = BackgroundScheduler()

client = MongoClient(host=configs.MONGO\_HOST, port=configs.MONGO\_PORT)

\# collection为存储定时任务的文档(表)，database为数据库

store = MongoDBJobStore(collection='scheduler\_job', database='admin', client=client)

scheduler.add\_jobstore(store)

\# scheduler.add\_job(job, 'interval', seconds=5, id=str(time.time()))

return scheduler

使用

def schedule\_clear\_db():

"""定时任务清除数据库"""

\# 每周末晚上2点半删除过期数据(6个月之前的数据,这里的id对应mongodb中的\_id

\# day\_of\_week设置一周中的哪几天执行，使用前三位字母代替，args为参数，这里的6表示6个月，

\# timezone最好设置，linux时区一般为utc，del\_historical\_data为定时任务实际执行的用来删除数据的函数

schedule = get\_scheduler()

schedule.add\_job(del\_historical\_data, 'cron', day\_of\_week='sat-sun', hour=2, minute=30, id="db\_clear",

args=\[6\], replace\_existing=True, timezone="Asia/Shanghai")

if schedule.state == 0:

schedule.start()

在mongo中查看任务

使用图形化工具查看定时任务，这里我使用的是mongodb compass

![eac3b51711b4][]

定时任务信息

格式化数据库定时任务下次执行时间(next\_run\_time)

print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(1600453800)))

\# 2020-09-19 02:30:00

从结果定时任务启用成功，而且时间一致。

修改定时任务

实际中可能需要随时修改数据的保存时间，在实际web项目(django或者flask等)中加一个接口即可。

修改可以使用reschedule\_job、modify\_job、add\_job方法。结合实际使用即可，这里使用add\_job方法。

\# add\_job 如果id重复则会覆盖原来的任务

schedule.add\_job(del\_historical\_data, 'cron', day\_of\_week='sat-sun', hour=2, minute=40, id="db\_clear",

args=\[save\_months\], replace\_existing=True, timezone="Asia/Shanghai")

在flask中使用核心代码示例

app.py

from datetime import timedelta

from configuration import con, schedule\_clear\_db

from configs import \*

from flask\_cors import CORS

from flask import Flask

app = Flask(\_\_name\_\_)

CORS(app, supports\_credentials=True) \# 设置全局跨域

app.register\_blueprint(con, url\_prefix='configuration') \# 路由分发

app.config\['SEND\_FILE\_MAX\_AGE\_DEFAULT'\] = timedelta(seconds=1) \# 缓存时间为1秒

if \_\_name\_\_ == '\_\_main\_\_':

schedule\_clear\_db()

app.run(host='0.0.0.0', port="5000", debug=False, threaded=True)

init.py

from flask import Blueprint

con = Blueprint("configuration", \_\_name\_\_)

from configuration.views import \*

views.py

import flask

from configuration import con

from configuration.impl import \*

@con.route("/updateDbSaveTime", methods=\["POST"\])

def update\_clear\_db\_time\_inter():

"""修改清理过期数据时间设置，数据保留默认时间为6个月"""

data = json.loads(flask.request.get\_data(as\_text=True))

save\_months = data.get("save\_months")

return update\_clear\_db\_time(save\_months)

接口实现类impl.py

def update\_clear\_db\_time(save\_months):

"""定时任务清除数据库"""

\# 每周末晚上2点半删除过期数据(6个月之前的数据)

\# add\_job 如果id重复则会覆盖原来的任务

schedule.add\_job(del\_historical\_data, 'cron', day\_of\_week='sat-sun', hour=2, minute=40, id="db\_clear",

args=\[save\_months\], replace\_existing=True, timezone="Asia/Shanghai")

if schedule.state == 0:

schedule.start()

conn = PyMongoDB()

find\_one = conn.find\_one\_universal(col\_name="scheduler\_job", match=\{"\_id": "db\_clear"\})

conn.close()

next\_run\_time = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(find\_one.get("next\_run\_time")))

return \{"code": "00", "msg": "success", "next\_run\_time": next\_run\_time\}

接口测试

![eac3b51711b4][]

测试结果

[eac3b51711b4]: /images/20221024/c82d584a93984868a2ef99f223f31d1f.png

python apscheduler 堵塞_python使用Apscheduler定时清理数据

发表评论取消回复

还没有评论，来说两句吧...

相关阅读

相关 Python 定时任务框架 APScheduler 详解

相关 python使用corn表达式实现定时任务apscheduler

相关 python使用apscheduler库实现定时任务功能

相关【python】使用APScheduler或Django-apscheduler实现定时任务

相关 python apscheduler 堵塞_python使用Apscheduler定时清理数据

相关 Python 定时任务框架 apscheduler

相关 python定时任务调度——apscheduler模块

相关 Python定时任务调度——APScheduler

相关 python中apscheduler定时模块使用

相关 APschedule定时任务