APScheduler是一款功能非常強大的定時任務框架。利用APScheduler框架我們可以很方便實現一個基於Python的定時任務系統

1，安裝

pip install apscheduler

2，簡單呼叫（一次性呼叫任務）

from apscheduler.schedulers.blocking import BlockingScheduler
import datetime
def aps_test():
    print (datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), '你好')

scheduler = BlockingScheduler()
scheduler.add_job(func=aps_test, next_run_time=datetime.datetime.now() + datetime.timedelta(seconds=2))
scheduler.start()
 

啟動後兩秒，執行任務aps_test()

3，在flask中使用APScheduler（也是一次呼叫性任務）

from flask import Flask,jsonify
from flask_apscheduler import APScheduler
import datetime
import random

app = Flask(__name__)
queue = []

class Config(object):
    JOBS = [
        {
            'id': 'jobjob',
            # 'func': 'jobs:job1',
            'func': '__main__:job1',
            'args': (1, 2),
            'next_run_time':datetime.datetime.now() + datetime.timedelta(seconds=10)
        }
    ]
    SCHEDULER_API_ENABLED = True

def job1(a, b):
    global queue
    while True:
        if len(queue) > 10:
            queue = []
        else:
            ranInfo = random.randint(1, 7)  # 隨機1-7
            queue.append(ranInfo)

@app.route("/getImgInfo")
def getImgInfo():
    global queue
    return jsonify({'status': 0, 'errmsg': queue})

if __name__ == '__main__':
    app.config.from_object(Config())
    scheduler = APScheduler()
    # it is also possible to enable the API directly
    # scheduler.api_enabled = True
    scheduler.init_app(app)
    scheduler.start()
    app.run()
 

說明：使用定時排程任務是為了解決一個問題。因為後臺有個死迴圈程式，相當於上面的job1，我要獲得該程式中的queue資訊。不使用排程任務，web程式就啟動不起來。我想過使用訊息佇列，把web程式和死迴圈程式拆成兩個程式。一個向訊息佇列中發訊息，另一個向訊息佇列中取訊息。但那樣太麻煩了，還有許多這樣那樣的問題。

現在使用定時排程任務就能完美解決了，上面的程式是在啟動後10秒這行job1任務。

4，參考部落格地址