Python使用apscheduler模組設定定時任務的實現

2022-05-25 22:01:55

一、安裝

pip install apscheduler

二、ApScheduler 簡介

1 APScheduler的元件

triggers：觸發器
triggers包含任務執行的排程邏輯，決定任務按照什麼邏輯進行定時執行

job stores；任務記憶體
儲存了排程任務

executors：執行器
用例執行任務的，包含執行緒池以及程序池等的建立和呼叫等等

schedulers：排程器
屬於控制面，將其他幾個方面組織起來的作用、

2 排程器的種類

排程器有以下幾種常見型別，其中最常用的BackgroundScheduler，即非阻塞式，因為在一般情況下，定時任務都會在放到web服務中，如果使用阻塞式，則無法啟動web服務，而使用非阻塞式，則將定時任務設定後，就不管了，繼續執行後面的web服務，只要web服務在執行，定時任務就是一直有效的

BlockingScheduler: 阻塞式
BackgroundScheduler: 非阻塞式（後臺執行）
AsyncIOScheduler: 當使用asyncio模組時使用
GeventScheduler: 當使用gevent模組時使用
TornadoScheduler: 構建Tornado應用時使用
TwistedScheduler: 構建Twisted應用時使用
QtScheduler: 構建Qt應用時使用

3 內建的觸發器型別

date: 在某個時間點執行一次時使用
interval: 固定的時間間隔迴圈執行時使用
cron: 在一天中特定的時間點執行時使用
calendarinterval: 當想在在一天中特定時間點或以日曆為基礎的時間間隔內執行時使用

三、使用舉例

這裡jiu就以非阻塞式BackgroundScheduler排程器為例展開

1 使用date型別的觸發器

如下，使用了三種設定日期和時間的方法

from apscheduler.schedulers.background import BackgroundScheduler
import time
from datetime import date
from datetime import datetime


def do_func(name,age):
    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+" in do func : 姓名："+name+" 年齡："+str(age))

def main():
    print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    sched=BackgroundScheduler()

    # 通過date 設定指定日期執行
    sched.add_job(do_func,trigger="date",run_date=date(2022,5,25),args=("張三丰",100))

    # 通過datetime，設定指定日期額指定時刻執行
    sched.add_job(do_func, trigger="date", run_date=datetime(2022, 5, 25,14,0,10), args=("張三丰", 100))

    # 直接使用文字的方式指定日期和時刻表
    sched.add_job(do_func, trigger="date", run_date="2022-05-25 14:0:20", args=("張三丰", 100))

    sched.start()

if __name__=="__main__":
    main()
    while True:
        print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
        time.sleep(1)

執行結果如下,可以發現，第一個通過date指定日期的預設是0點執行，顯然時間已經過了，不會執行，第二和第三個則在規定的時間點執行了，這裡還需要注意的是，通過列印可以看出，main函數執行完成後，已經開始執行main函數下面的while迴圈列印語句了，而在執行迴圈的過程中，定時任務仍然生效，這就是非阻塞式排程器的原理，如果是阻塞式，則在此程式碼中，會一直卡在main函數中，main下面額while迴圈語句是不會執行的，因此在實際使用中，非阻塞式應用的是非常多的

2022-05-25 14:00:02
2022-05-25 14:00:02
C:Usershitre.virtualenvszentaolinkgitlab-QCS5yxD9libsite-packagesapschedulerutil.py:436: PytzUsageWarning: The localize method is no longer necessary, as this time zone supports the fold attribute (PEP 495). For more details on migrating to a PEP 495-compliant implementation, see https://pytz-deprecation-shim.readthedocs.io/en/latest/migration.html
return tzinfo.localize(dt)
Run time of job "do_func (trigger: date[2022-05-25 00:00:00 CST], next run at: 2022-05-25 00:00:00 CST)" was missed by 14:00:02.088260
2022-05-25 14:00:03
2022-05-25 14:00:04
2022-05-25 14:00:05
2022-05-25 14:00:06
2022-05-25 14:00:07
2022-05-25 14:00:08
2022-05-25 14:00:09
2022-05-25 14:00:10 in do func : 姓名：張三丰年齡：100
2022-05-25 14:00:10
2022-05-25 14:00:11
2022-05-25 14:00:12
2022-05-25 14:00:13
2022-05-25 14:00:14
2022-05-25 14:00:15
2022-05-25 14:00:16
2022-05-25 14:00:17
2022-05-25 14:00:18
2022-05-25 14:00:19
2022-05-25 14:00:20 in do func : 姓名：張三丰年齡：100
2022-05-25 14:00:20
2022-05-25 14:00:21
2022-05-25 14:00:22

2 使用interval型別的觸發器

如下程式碼演示了時間間隔迴圈執行的使用例子

from apscheduler.schedulers.background import BackgroundScheduler
import time
from datetime import date
from datetime import datetime


def do_func(name,age):
    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+" in do func : 姓名："+name+" 年齡："+str(age))

def main():
    print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    sched=BackgroundScheduler()

    # 每3秒執行一次
    sched.add_job(do_func,trigger="interval",args=("張三丰",100),seconds=3)
    # 每3分鐘執行一次
    sched.add_job(do_func, trigger="interval", args=("張三丰", 100), minutes=3)
    # 每3小時執行一次
    sched.add_job(do_func, trigger="interval", args=("張三丰", 100), hours=3)
    # 每3天執行一次
    sched.add_job(do_func, trigger="interval", args=("張三丰", 100), days=3)
    # 每3周執行一次
    sched.add_job(do_func, trigger="interval", args=("張三丰", 100), weeks=3)

    sched.start()

if __name__=="__main__":
    main()
    while True:
        print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
        time.sleep(1)

上面的程式碼中因為時間跨度比較大，這裡只演示妹3秒執行一次的程式碼

程式碼如下：

from apscheduler.schedulers.background import BackgroundScheduler
import time
from datetime import date
from datetime import datetime


def do_func(name,age):
    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+" in do func : 姓名："+name+" 年齡："+str(age))

def main():
    print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    sched=BackgroundScheduler()

    # 每3秒執行一次
    sched.add_job(do_func,trigger="interval",args=("張三丰",100),seconds=3)

    sched.start()

if __name__=="__main__":
    main()
    while True:
        print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
        time.sleep(1)

執行結果如下：

2022-05-25 14:14:04
2022-05-25 14:14:04
2022-05-25 14:14:05
2022-05-25 14:14:06
2022-05-25 14:14:07 in do func : 姓名：張三丰年齡：100
2022-05-25 14:14:07
2022-05-25 14:14:08
2022-05-25 14:14:09
2022-05-25 14:14:10 in do func : 姓名：張三丰年齡：100
2022-05-25 14:14:10
2022-05-25 14:14:11
2022-05-25 14:14:12
2022-05-25 14:14:13 in do func : 姓名：張三丰年齡：100
2022-05-25 14:14:13

3 使用cron型別的觸發器

cron的觸發器有點類似linux上crontab定時器的使用，程式碼如下：

from apscheduler.schedulers.background import BackgroundScheduler
import time
from datetime import date
from datetime import datetime


def do_func(name,age):
    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+" in do func : 姓名："+name+" 年齡："+str(age))

def main():
    print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    sched=BackgroundScheduler()

    # 任務會在6月、7月、8月、11月和12月的第三個週五，00:00、01:00、02:00和03:00觸發
    sched.add_job(do_func,trigger="cron",month='6-8,11-12', day='3rd fri', hour='0-3',args=("張三丰",100))

    sched.start()

if __name__=="__main__":
    main()
    while True:
        print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
        time.sleep(1)

這裡需要注意的是，可以省略不需要的欄位。當省略時間引數時，在顯式指定引數之前的引數會被設定為*，之後的引數會被設定為最小值，week 和day_of_week的最小值為*

day=1, minute=20
等同於
year='*', month='*', day=1, week='*', day_of_week='*', hour='*', minute=20, second=0

此外，也可以直接使用crontab表示式，如下：

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
import time
from datetime import date
from datetime import datetime


def do_func(name,age):
    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+" in do func : 姓名："+name+" 年齡："+str(age))

def main():
    print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    sched=BackgroundScheduler()

    # 任務會在6月、7月、8月、11月和12月的第三個週五，00:00、01:00、02:00和03:00觸發
    sched.add_job(do_func,trigger=CronTrigger.from_crontab('48 10 1-15 sep-nov *'),args=("張三丰",100))

    sched.start()

if __name__=="__main__":
    main()
    while True:
        print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
        time.sleep(1)

四、定時器使用裝飾器的方法

以間隔時間迴圈執行的程式碼為例，如下為未使用裝飾器的方式

from apscheduler.schedulers.background import BackgroundScheduler
import time
from datetime import date
from datetime import datetime


def do_func(name,age):
    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+" in do func : 姓名："+name+" 年齡："+str(age))

def main():
    print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    sched=BackgroundScheduler()

    # 每3秒執行一次
    sched.add_job(do_func,trigger="interval",args=("張三丰",100),seconds=3)

    sched.start()

if __name__=="__main__":
    main()
    while True:
        print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
        time.sleep(1)

修改為使用裝飾器的方式如下：

from apscheduler.schedulers.background import BackgroundScheduler
import time

sched=BackgroundScheduler()

@sched.scheduled_job(trigger="interval",args=("張三丰",100),seconds=3)
def do_func(name,age):
    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+" in do func : 姓名："+name+" 年齡："+str(age))

if __name__=="__main__":
    sched.start()
    while True:
        print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
        time.sleep(1)