flask-定时任务

---

前言

有时候我们需要在服务器做一些定时任务，虽然可以使用crontab去加一些定时任务，但是python自己就有定时任务为啥不消起来，下面就是怎么简单使用apscheduler去创建python的定时任务调度，仅供参考

---

一、APScheduler是什么

APScheduler（全名 Advanced Python Scheduler）是一个功能强大的 Python 库，用于在应用步伐中调度和实行定时任务。它允许你在一定时间间隔、特定日期和时间或特定条件下主动实行某些任务。APScheduler 可以在差别的情况中使用，如 Flask、Django 或其他 Python 应用
二、APScheduler 主要功能：

• 定时任务调度：你可以设置任务在指定的时间间隔、固定时间、或特定日期/时间点实行。
  
• 差别范例的触发器：如时间间隔触发、特定时间触发等。
  
• 任务存储：APScheduler 支持将调度的任务存储在数据库中，保持任务状态，纵然应用步伐重启后也能恢复任务。
  
• 背景调度器：它允许任务在背景运行，不会影响主应用步伐的正常运行。
  

三、主要组成部分：

• Scheduler（调度器）：调度器是管理任务和调度实行的焦点。它根据触发器的设置来调度任务实行。
  
• Job（任务）：一个被调度的任务，是具体实行的功能或使用。
  
• Trigger（触发器）：触发器决定了任务何时实行，通常有几种范例：
  
• interval：按间隔实行任务（如每 5 分钟实行一次）。
  
• cron：在特定时间或日期实行任务（比方每天破晓 1 点）。
  
• date：在指定的某个时间点实行任务。
  
• Executors（实行器）：实行器负责实行任务。它定义了如何启动任务的工作线程或进程。
  

四、典型使用场景：

• 定时清算数据：定期实行某些清算任务，比方每晚清算日志。
  
• 定时发送通知：按固定时间发送邮件或消息。
  
• 定时抓取数据：定期从 API 抓取数据，实行定时的数据同步使用等
  

五、具体使用

1.安装 APScheduler

1. pip install apscheduler

复制代码

大概写到requirements.txt然后实行

1. pip install -r requirements.txt

复制代码

2.假设我们有一个需要五分钟请求一次http接口的任务

1.定义一个scheduler.py去专门处理定时

1. # scheduler.py
2. from apscheduler.schedulers.background import BackgroundScheduler
3. import time
4. import requests
6. def fetch_material_info():
7.     print("定时任务开始执行")
8.     try:
9.         response = requests.post('http://127.0.0.1:5002/api/material/get_material_info_by_api')#这个api可以是自身接口也可以是第三方接口
10.         if response.status_code == 200:
11.             print("成功获取材料信息")
12.         else:
13.             print(f"请求失败，状态码：{response.status_code}")
14.     except Exception as e:
15.         print(f"请求失败: {e}")
17. def start_scheduler():
18.     print("启动定时任务")
19.     scheduler = BackgroundScheduler()
20.     scheduler.add_job(fetch_material_info, 'interval', minutes=5)
21.     scheduler.start()
22.     print("定时任务启动成功")
24.     # 防止 APScheduler 被杀死，保持主线程活着
25.     try:
26.         while True:
27.             time.sleep(10)
28.     except (KeyboardInterrupt, SystemExit):
29.         scheduler.shutdown()
31. if __name__ == '__main__':
32.     start_scheduler()

复制代码

2.启动文件处理

1. from app import create_app
2. import subprocess
4. # 创建 Flask 应用
5. app = create_app()
7. # 启动定时任务
8. def start_scheduler():
9.         # 这里的print都可以写到日志里面，配置一下logging就可以了
10.     print("启动定时任务...")
11.     # 使用 subprocess.Popen 启动定时任务
12.     process = subprocess.Popen(['python', 'scheduler.py'])
13.     print("定时任务已启动, PID: {}".format(process.pid))
15. if __name__ == "__main__":
16.     # 启动定时任务
17.     start_scheduler()
19.     # 启动 Flask 应用
20.     app.run(host="0.0.0.0", port=5002, debug=True)

复制代码

3.使用docker启动引入守护进程

1. [supervisord]
2. nodaemon=true
4. [program:flask]
5. command=python wsgi.py
6. autostart=true
7. autorestart=true
8. stderr_logfile=/var/log/flask.err.log
9. stdout_logfile=/var/log/flask.out.log
11. [program:scheduler]
12. command=python scheduler.py
13. autostart=true
14. autorestart=true
15. stderr_logfile=/var/log/scheduler.err.log
16. stdout_logfile=/var/log/scheduler.out.log

复制代码

4.docker镜像示例（无法使用docker-compose以是使用docker镜像）

1. # 使用 Python 3.9 的官方镜像作为基础镜像FROM python:3.9# 设置工作目次WORKDIR /home/app# 将当前目次下的所有文件复制到镜像的工作目次COPY . /home/app/# 创建一个虚拟情况RUN python -m venv venv# 激活虚拟情况并安装项目标依靠RUN ./venv/bin/pip install --upgrade pipRUN ./venv/bin/pip install -r requirements.txt
2. # 设置情况变量，使用虚拟情况ENV PATH="/home/app/venv/bin:$PATH"# 安装 Gunicorn（如果尚未包罗在 requirements.txt 中）RUN ./venv/bin/pip install gunicorn# 安装 Supervisor 来管理多个进程RUN apt-get update && apt-get install -y supervisor# 将 supervisor 设置文件复制到容器中COPY supervisor.conf /etc/supervisor/conf.d/supervisor.conf# 使用 Supervisor 启动 Flask 应用和定时任务CMD ["supervisord", "-c", "/etc/supervisor/supervisord.conf"]

复制代码

5.镜像启动之后的结果

总结

APScheduler 是一个强大的定时任务调度库，实用于需要定期实行任务的 Python 应用步伐。它提供了机动的调度选项，支持多种范例的触发器，非常得当在背景任务、主动化任务等场景中使用。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。