简介:APScheduler是一个轻量级的Python库,用于在后台运行定时任务和延迟任务。它可以轻松地安排任务并支持多种类型的触发器,例如固定间隔、日期/时间表达式、CRON表达式等。APScheduler还提供了多个后台调度器实现,例如基于线程池的调度器、基于进程池的调度器以及异步调度器。
历史攻略:
Python:Celery+Redis+Flower安装和使用
Python:Celery+Redis实现定时任务
定时任务:Python
安装依赖库:
pip install apscheduler flask
案例源码:
# -*- coding: utf-8 -*-
# time: 2023/4/26 11:25
# file: main.py
# 公众号: 玩转测试开发
from flask import Flask, request
from apscheduler.schedulers.background import BackgroundScheduler
import requests
app = Flask(__name__)
def send_message():
url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=your-key"
data = {
"msgtype": "text",
"text": {"content": "Hello, World!"}
}
r = requests.post(url, json=data)
scheduler = BackgroundScheduler()
@app.route("/send_message")
def schedule_message():
minutes = int(request.args.get("minutes", 1))
job_id = f"send_message_{minutes}"
scheduler.add_job(func=send_message, trigger="interval", minutes=minutes, id=job_id)
return f"Message scheduled to be sent every {minutes} minutes with job ID: {job_id}"
@app.route("/delete_message")
def delete_message():
job_id = request.args.get("job_id")
if job_id:
try:
scheduler.remove_job(job_id)
return f"Job {job_id} deleted."
except Exception as e:
return f"Error deleting job {job_id}: {e}"
else:
return "Please provide a valid job ID."
if __name__ == "__main__":
scheduler.start()
app.run(debug=True)
运行结果:
推送的结果:
注意事项:
APScheduler是一个后台任务调度库,应该在主线程之外运行。
APScheduler支持多种类型的触发器,例如固定间隔、日期/时间表达式、CRON表达式等。请根据您的需求选择合适的触发器。
如果您需要在多个进程中使用APScheduler,请考虑使用基于数据库的调度器,例如SQLAlchemyJobStore或MongoDBJobStore。
APScheduler提供了许多可自定义的选项和回调函数,例如任务完成时要运行的回调函数。请查看文档以了解更多信息。
为防止任务重复执行,请确保设置唯一的ID参数。