标签：fcntl 启动 Flask app ApScheduler 任务 scheduler flask

Flask-ApScheduler 任务未执行问题

在使用flask run启动 flask 时，任务没有按计划执行

ApScheduler 任务配置

Config = {
    JOBS = [
        {
            'id': 'ims_info',  # 标识
            'func': 'task_1:main',  # 运行函数
            'trigger': 'interval',  # 定时任务的类型
            'seconds': 900  # 运行的间隔时间
        }
    ]
    SCHEDULER_API_ENABLED = True
    SCHEDULER_TIMEZONE = 'Asia/Shanghai'
    SCHEDULER_LOCK_FILE = 'scheduler.lock'
}

ApScheduler 初始化

为了防止使用 gunicorn 启动 flask 时，重复启动任务，在创建 Apscheduler 对象时增加文件锁

import fcntl
import atexit
import flask
from flask import Flask, request, g, make_response
from flask_apscheduler import APScheduler
from apscheduler.schedulers.background import BackgroundScheduler


def create_apscheduler(flask_app):
    """创建 Apscheduler 对象，使用文件锁防止用 gunicorn 启动 flask 时，重复启动多个任务"""
    f = open(Config.SCHEDULER_LOCK_FILE, 'wb')
    try:
        fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
        scheduler = APScheduler(BackgroundScheduler())  # 创建 Apscheduler 对象
        scheduler.init_app(flask_app)
        scheduler.start()  # 启动任务列表
    except Exception as _:
        pass

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

    atexit.register(unlock)


app = Flask(__name__)
app.config.from_object(Config)
create_apscheduler(app)

问题原因

flask 使用 debug 模式启动时，会创建一个子分支，Flask-ApScheduler 为了防止启动两次任务，只在子进程中启动任务。

# Flask-ApScheduler 启动源码
def start(self, paused=False):
    """
    Start the scheduler.
    :param bool paused: if True, don't start job processing until resume is called.
    """

    # Flask in debug mode spawns a child process so that it can restart the process each time your code changes,
    # the new child process initializes and starts a new APScheduler causing the jobs to run twice.
    if flask.helpers.get_debug_flag() and not werkzeug.serving.is_running_from_reloader():
        return

    if self.host_name not in self.allowed_hosts and '*' not in self.allowed_hosts:
        LOGGER.debug('Host name %s is not allowed to start the APScheduler. Servers allowed: %s' %
                        (self.host_name, ','.join(self.allowed_hosts)))
        return

    self._scheduler.start(paused=paused)

这与前面防止 gunicorn 启动多次任务的文件锁机制冲突，最后导致任务一次也没有启动：

• 文件锁保证只在第一个进程中执行 Flask-ApScheduler 初始化及启动，也就是 debug 时的主进程；
• Flask-ApScheduler 的 start 方法中只会在子进程启动时启动任务；

问题解决

在文件锁前增加 debug 模式判断，当 debug 模式时，不使用文件锁：

def create_apscheduler(flask_app):
    """创建 Apscheduler 对象，使用文件锁防止用 gunicorn 启动 flask 时，重复启动多个任务"""
    if flask.helpers.get_debug_flag():
        scheduler = APScheduler(BackgroundScheduler())  # 创建 Apscheduler 对象
        scheduler.init_app(flask_app)
        scheduler.start()  # 启动任务列表
    else:
        f = open(Config.SCHEDULER_LOCK_FILE, 'wb')
        try:
            fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
            scheduler = APScheduler(BackgroundScheduler())  # 创建 Apscheduler 对象
            scheduler.init_app(flask_app)
            scheduler.start()  # 启动任务列表
        except Exception as _:
            pass

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

        atexit.register(unlock)

标签：fcntl,启动,Flask,app,ApScheduler,任务,scheduler,flask
From： https://www.cnblogs.com/shouwangrenjian/p/16874760.html