APScheduler的全稱是Advanced Python Scheduler。它是一個輕量級的 Python 定時任務調度框架。APScheduler 支持三種調度任務：固定時間間隔，固定時間點（日期），Linux 下的 Crontab 命令。同時，它還支持異步執行、后台執行調度任務。官方文檔
Flask與APscheduler結合
-
安裝
pip install flask-apscheduler -
將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