1、背景

最近需要做拼音搜索的定时更新功能。因此考虑在拼音搜索的接口里加入定时调度功能，以定时更新拼音搜索的语料库。
但是，采用Flask-APScheduler模块时，在本地测试出现了后台被重复执行了2次的情况。

2、异常项目代码

api_flask.py文件

 # api_flask.py
import json
import datetime

from flask_apscheduler import APScheduler
from flask import Flask, request, Response

from api_schedule.calculate import calcu,write
from api_schedule import main

app = Flask(__name__)
num = 0


class Config(object):
    JOBS = [
        {
            'id': '100',  # 不重复的标识
            'func': 'api_flask:scheduler_job',  # 定时执行的 模块：函数
            'trigger': 'interval',              # 定时执行，其他可选参数data,interval
            'seconds': 3
            # 'trigger': 'cron',                  # 定时执行，其他可选参数data,interval
            # 'hour': 3,                          # 每天3：00分执行
            # 'minute': 0
        }
    ]
    SCHEDULER_API_ENABLED = True


def scheduler_job():  # 一个函数，用来做定时任务的任务。
    global num
    write()
    num += 1
    print("后台执行完毕，执行次数num = {}, time = {}".format(num, datetime.datetime.now()))

    import importlib
    importlib.reload(main)


@app.route('/', methods=['GET', 'POST'])
def api_test():
    res = main.main_process()
    res_json = json.dumps({'data':res})
    print("return res = {}".format(res_json))

    return Response(res_json, mimetype='application/json')


def scheduler_init(app):
    scheduler = APScheduler()
    scheduler.init_app(app)
    scheduler.start()
    print("scheduler started.")

app.config.from_object(Config)
scheduler_init(app)


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080, debug=False)

calculate.py文件

#calculate.py
def calcu():
    with open('./data.txt', 'r') as f:
        res = f.read()
    return res

def write():
    res = int(calcu())
    with open('./data.txt', 'w', encoding='utf-8') as f:
        # data.txt里写入一个整数即可。
        f.writelines(str(res+1))
if __name__ == '__main__':
    print(calcu())
    write()
    print(calcu())

3、异常日志

D:\Users\Leslie\Anaconda3\python.exe F:/codes/study/api_schedule/api_flask.py
scheduler started.
scheduler started.
 * Serving Flask app "api_flask" (lazy loading)
 * Environment: production
   WARNING: This is a development server. Do not use it in a production deployment.
   Use a production WSGI server instead.
 * Debug mode: off
 * Running on http://0.0.0.0:8080/ (Press CTRL+C to quit)
后台执行完毕，执行次数num = 1, time = 2021-01-05 16:17:02.920408
后台执行完毕，执行次数num = 2, time = 2021-01-05 16:17:02.926411
后台执行完毕，执行次数num = 3, time = 2021-01-05 16:17:05.921030
后台执行完毕，执行次数num = 4, time = 2021-01-05 16:17:05.926030

Process finished with exit code -1

4、解决办法

将定时调度模块的代码放入main函数下，而不是放在上面的主代码块。具体原因还没搞清楚。
即将下面的代码放在 if name == 'main':之后。

注意：app.run(host='0.0.0.0', port=8080, debug=False)中必须设置debug=False，否则会出现另外一种重复执行2次的情况。

def scheduler_init(app):
    scheduler = APScheduler()
    scheduler.init_app(app)
    scheduler.start()
    print("scheduler started.")

app.config.from_object(Config)
scheduler_init(app)

调整后的代码：

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# Author：Leslie Dang
# File:   main.py
# Initial Data : 2020/12/11 16:01

import json
import datetime

from flask_apscheduler import APScheduler
from flask import Flask, request, Response

from api_schedule.calculate import calcu,write
from api_schedule import main

app = Flask(__name__)
num = 0


class Config(object):
    JOBS = [
        {
            'id': '100',  # 不重复的标识
            'func': 'api_flask:scheduler_job',  # 定时执行的 模块：函数
            'trigger': 'interval',              # 定时执行，其他可选参数data,interval
            'seconds': 3
            # 'trigger': 'cron',                  # 定时执行，其他可选参数data,interval
            # 'hour': 3,                          # 每天3：00分执行
            # 'minute': 0
        }
    ]
    SCHEDULER_API_ENABLED = True


def scheduler_job():  # 一个函数，用来做定时任务的任务。
    global num
    write()
    num += 1
    print("后台执行完毕，执行次数num = {}, time = {}".format(num, datetime.datetime.now()))

    import importlib
    importlib.reload(main)


@app.route('/', methods=['GET', 'POST'])
def api_test():
    res = main.main_process()
    res_json = json.dumps({'data':res})
    print("return res = {}".format(res_json))

    return Response(res_json, mimetype='application/json')


if __name__ == '__main__':
    def scheduler_init(app):
        scheduler = APScheduler()
        scheduler.init_app(app)
        scheduler.start()
        print("scheduler started.")

    app.config.from_object(Config)
    scheduler_init(app)

    app.run(host='0.0.0.0', port=8080, debug=False)

调整后的运行日志

D:\Users\Leslie\Anaconda3\python.exe F:/codes/study/api_schedule/api_flask.py
scheduler started.
 * Serving Flask app "api_flask" (lazy loading)
 * Environment: production
   WARNING: This is a development server. Do not use it in a production deployment.
   Use a production WSGI server instead.
 * Debug mode: off
 * Running on http://0.0.0.0:8080/ (Press CTRL+C to quit)
后台执行完毕，执行次数num = 1, time = 2021-01-05 16:26:29.704507
后台执行完毕，执行次数num = 2, time = 2021-01-05 16:26:32.705826
后台执行完毕，执行次数num = 3, time = 2021-01-05 16:26:35.705605

Process finished with exit code -1

Flask-APScheduler重复执行2次的解决办法