您现在的位置是：首页 > Python

当前栏目

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

Python 数据库数据库数据

2023-03-02 10:56:14 时间

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

猜你喜欢

如何在苹果appstore上架应用
图形编辑器：修改图形X、Y、Width、Height、Rotation
「拯救地球」实验成功！小行星偏离轨道，运行周期缩短32分钟
面试官：请求转发和请求重定向有什么区别？
得物容器SRE探索与实践
Numpy中数组和矩阵操作的数学函数
软件测试|SonarQube 安装、配置及 JaCoCo、Maven 集成
基于不同操作系统升级知行之桥的常见问题
后端思维篇：如何抽取一个观察者模板
四行代码秒解微积分！Python 这个模块神了！
99%的人都不知道！Python、C、C 扩展、Cython 差异对比！
【每日要闻】台积电三季度净利大涨79.7%；RTX 4090首发即遭黄牛疯炒
大众牵手地平线，下一步反攻特斯拉？
仙途智能完成B3轮战略融资，商汤国香资本等机构加持｜镁客网每周硬科技领域投融资汇总（10.8-10.14）
ONNXRUNTIEM版本升级到1.13的大坑
Pandas处理数据太慢，来试试Polars吧！
如果后端API一次返回10万条数据，前端应该如何处理？
轻松学Pytorch之Deeplabv3推理
一行Pandas代码制作数据分析透视表，太牛了！
八个无需编写代码即可使用 Python 内置库的方法

zl程序教程

当前栏目

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

相关文章