[Flask]Flask结合APScheduler实现定时任务框架

APScheduler的全称是Advanced Python Scheduler。它是一个轻量级的 Python 定时任务调度框架。APScheduler 支持三种调度任务：固定时间间隔，固定时间点（日期），Linux 下的 Crontab 命令。同时，它还支持异步执行、后台执行调度任务。官方文档

---

Flask与APscheduler结合

1. 安装pip install flask-apscheduler

2. 将apscheduler注册到Flask App

   from flask_apscheduler import APScheduler
   
   app = Flask(__name__)
   scheduler = APScheduler()
   app.config.update(
       {
           "SCHEDULER_API_ENABLED": True,
           "SCHEDULER_TIMEZONE": "Asia/Shanghai",
           "JOBS": 
           [
               {
                   "id": "job1",   # 任务ID
                   "func": "admin:tec.job1",   # 任务位置
                   "trigger": "interval",   # 触发器
                   "seconds": 5   # 时间间隔
               },
               {
                   "id": "job2",   # 任务ID
                   "func": "admin:tec.job2",   # 任务位置
                   "trigger": "cron",   # cron触发器
                   "day_of_week": "mon-sun",   # 星期一到星期天每天7点10分5秒执行
                   "hour": 7,
                   "minute": 10,
                   "second": 5   
               }
           ]
       }
   )
   
   scheduler.init_app(app)
   scheduler.start()
   

---

踩坑点

定时任务重复启动

解决方法

在启动任务时，设置文件锁，当能获取到文件锁时，不在启动任务

# 保证apschedule只执行一次
if platform.system() != 'Windows':
    fcntl = __import__("fcntl")
    f = open('scheduler.lock', 'wb')
    try:
        fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
        scheduler.init_app(app)
        scheduler.start()
        app.logger.debug('Scheduler Started,---------------')
    except:
        pass

    def unlock():
        fcntl.flock(f, fcntl.LOCK_UN)
        f.close()

    atexit.register(unlock)
else:
    msvcrt = __import__('msvcrt')
    f = open('scheduler.lock', 'wb')
    try:
        msvcrt.locking(f.fileno(), msvcrt.LK_NBLCK, 1)
        scheduler.init_app(app)
        scheduler.start()
        app.logger.debug('Scheduler Started,----------------')
    except:
        pass

    def _unlock_file():
        try:
            f.seek(0)
            msvcrt.locking(f.fileno(), msvcrt.LK_UNLCK, 1)
        except:
            pass

    atexit.register(_unlock_file)

---

找不到func所配置的model

在实际项目中，定时任务的函数不可能都放在一个文件夹下，app.config的func属性即指明任务的位置

解决方法

如任务在admin.tec路径下的card函数，需要配置为"func": "admin:tec.card"

---

在任务函数中使用SQLAlchemy报错

报错No application found. Either work inside a view function or push an application context，发现只能在视图函数中使用数据库操作，其他函数中使用数据库操作则报错，原因是上下文不完整

解决方法

• 方式一 直接实例化 db = SQLAlchemy(app) 或者 db = SQLAlchemy() db.create_all(app=create_app())

• 方式二 在函数中加上wtih app.app_context():

  def my_function:
  	with app.app_context():
  		user = db.User(...)
  		db.session.add(user)
  		db.session.commit()
  

• 方式三 创建app实例后，推送app程序的上下文app.app_context().push()

---

部署到服务器上Timezone出错

报错Timezone offset does not match system offset: 0 != 28800. Please, check your config files.

解决方法

• 在实例化APscheduler时加上时区，如：scheduler = APScheduler(BackgroundScheduler(timezone="Asia/Shanghai"))

然后在配置文件app.config中添加SCHEDULER_TIMEZONE = 'Asia/Shanghai'

---

Gunicorn使用gevent模式无效

解决方法

将gunicorn启动模式换为eventlet，配置文件gun.conf

# 并行工作进程数
workers = 4
# 指定每个工作者的线程数
threads = 4
# 监听内网端口80
bind = '0.0.0.0:80'
# 工作模式协程
worker_class = 'eventlet' 
# 设置最大并发量
worker_connections = 2000
# 设置进程文件目录
pidfile = 'gunicorn.pid'
# 设置访问日志和错误信息日志路径
accesslog = './logs/gunicorn_acess.log'
errorlog = './logs/gunicorn_error.log'
# 设置日志记录水平
loglevel = 'info'
# 代码发生变化是否自动重启
reload=True

---

总结

相关项目

• LeonAuto自动化系统

参考链接

• Flask后端实践 连载九 Flask-APScheduler定时任务与坑点解决方法

• Flask结合APScheduler实现定时任务框架

• flask_apscheduler 定时任务踩坑记录

• flask_apscheduler 定时任务框架