flask-定时任务

---

前言

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

---

一、APScheduler是什么

APScheduler（全名 Advanced Python Scheduler）是一个功能强大的 Python 库，用于在应用程序中调度和执行定时任务。它允许你在一定时间间隔、特定日期和时间或特定条件下自动执行某些任务。APScheduler 可以在不同的环境中使用，如 Flask、Django 或其他 Python 应用

二、APScheduler 主要功能：

1. 定时任务调度：你可以设置任务在指定的时间间隔、固定时间、或特定日期/时间点执行。
2. 不同类型的触发器：如时间间隔触发、特定时间触发等。
3. 任务存储：APScheduler 支持将调度的任务存储在数据库中，保持任务状态，即使应用程序重启后也能恢复任务。
4. 后台调度器：它允许任务在后台运行，不会影响主应用程序的正常运行。

三、主要组成部分：

1. Scheduler（调度器）：调度器是管理任务和调度执行的核心。它根据触发器的设置来调度任务执行。
2. Job（任务）：一个被调度的任务，是具体执行的功能或操作。
3. Trigger（触发器）：触发器决定了任务何时执行，通常有几种类型：
4. interval：按间隔执行任务（如每 5 分钟执行一次）。
5. cron：在特定时间或日期执行任务（例如每天凌晨 1 点）。
6. date：在指定的某个时间点执行任务。
7. Executors（执行器）：执行器负责执行任务。它定义了如何启动任务的工作线程或进程。

四、典型使用场景：

1. 定时清理数据：定期执行某些清理任务，例如每晚清理日志。
2. 定时发送通知：按固定时间发送邮件或消息。
3. 定时抓取数据：定期从 API 抓取数据，执行定时的数据同步操作等

五、具体使用

1.安装 APScheduler

pip install apscheduler

或者写到requirements.txt然后执行

pip install -r requirements.txt

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

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

# scheduler.py
from apscheduler.schedulers.background import BackgroundScheduler
import time
import requestsdef fetch_material_info():print("定时任务开始执行")try:response = requests.post('http://127.0.0.1:5002/api/material/get_material_info_by_api')#这个api可以是自身接口也可以是第三方接口if response.status_code == 200:print("成功获取材料信息")else:print(f"请求失败，状态码：{response.status_code}")except Exception as e:print(f"请求失败: {e}")def start_scheduler():print("启动定时任务")scheduler = BackgroundScheduler()scheduler.add_job(fetch_material_info, 'interval', minutes=5)scheduler.start()print("定时任务启动成功")# 防止 APScheduler 被杀死，保持主线程活着try:while True:time.sleep(10)except (KeyboardInterrupt, SystemExit):scheduler.shutdown()if __name__ == '__main__':start_scheduler()

2.启动文件处理

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

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

[supervisord]
nodaemon=true[program:flask]
command=python wsgi.py
autostart=true
autorestart=true
stderr_logfile=/var/log/flask.err.log
stdout_logfile=/var/log/flask.out.log[program:scheduler]
command=python scheduler.py
autostart=true
autorestart=true
stderr_logfile=/var/log/scheduler.err.log
stdout_logfile=/var/log/scheduler.out.log

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

# 使用 Python 3.9 的官方镜像作为基础镜像
FROM python:3.9# 设置工作目录
WORKDIR /home/app# 将当前目录下的所有文件复制到镜像的工作目录
COPY . /home/app/# 创建一个虚拟环境
RUN python -m venv venv# 激活虚拟环境并安装项目的依赖
RUN ./venv/bin/pip install --upgrade pip
RUN ./venv/bin/pip install -r requirements.txt# 设置环境变量，使用虚拟环境
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 应用程序。它提供了灵活的调度选项，支持多种类型的触发器，非常适合在后台任务、自动化任务等场景中使用。