Python使用apscheduler模塊設(shè)置定時(shí)任務(wù)的實(shí)現(xiàn)

更新時(shí)間：2022年05月25日 15:51:52 作者：redrose2100

本文主要介紹了Python使用apscheduler模塊設(shè)置定時(shí)任務(wù)的實(shí)現(xiàn)，文中通過示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下

一、安裝

pip install apscheduler

二、ApScheduler 簡(jiǎn)介

1 APScheduler的組件

triggers：觸發(fā)器
triggers包含任務(wù)執(zhí)行的調(diào)度邏輯，決定任務(wù)按照什么邏輯進(jìn)行定時(shí)執(zhí)行

job stores；任務(wù)存儲(chǔ)器
存儲(chǔ)了調(diào)度任務(wù)

executors：執(zhí)行器
用例執(zhí)行任務(wù)的，包含線程池以及進(jìn)程池等的創(chuàng)建和調(diào)用等等

schedulers：調(diào)度器
屬于控制面，將其他幾個(gè)方面組織起來的作用、

2 調(diào)度器的種類

調(diào)度器有以下幾種常見類型，其中最常用的BackgroundScheduler，即非阻塞式，因?yàn)樵谝话闱闆r下，定時(shí)任務(wù)都會(huì)在放到web服務(wù)中，如果使用阻塞式，則無法啟動(dòng)web服務(wù)，而使用非阻塞式，則將定時(shí)任務(wù)設(shè)定后，就不管了，繼續(xù)執(zhí)行后面的web服務(wù)，只要web服務(wù)在運(yùn)行，定時(shí)任務(wù)就是一直有效的

BlockingScheduler: 阻塞式
BackgroundScheduler: 非阻塞式（后臺(tái)運(yùn)行）
AsyncIOScheduler: 當(dāng)使用asyncio模塊時(shí)使用
GeventScheduler: 當(dāng)使用gevent模塊時(shí)使用
TornadoScheduler: 構(gòu)建Tornado應(yīng)用時(shí)使用
TwistedScheduler: 構(gòu)建Twisted應(yīng)用時(shí)使用
QtScheduler: 構(gòu)建Qt應(yīng)用時(shí)使用

3 內(nèi)置的觸發(fā)器類型

date: 在某個(gè)時(shí)間點(diǎn)執(zhí)行一次時(shí)使用
interval: 固定的時(shí)間間隔循環(huán)執(zhí)行時(shí)使用
cron: 在一天中特定的時(shí)間點(diǎn)執(zhí)行時(shí)使用
calendarinterval: 當(dāng)想在在一天中特定時(shí)間點(diǎn)或以日歷為基礎(chǔ)的時(shí)間間隔內(nèi)執(zhí)行時(shí)使用

三、使用舉例

這里jiu就以非阻塞式BackgroundScheduler調(diào)度器為例展開

1 使用date類型的觸發(fā)器

如下，使用了三種設(shè)置日期和時(shí)間的方法

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 設(shè)置指定日期執(zhí)行
? ? sched.add_job(do_func,trigger="date",run_date=date(2022,5,25),args=("張三豐",100))

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

? ? # 直接使用文本的方式指定日期和時(shí)刻表
? ? 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)

執(zhí)行結(jié)果如下,可以發(fā)現(xiàn)，第一個(gè)通過date指定日期的默認(rèn)是0點(diǎn)執(zhí)行，顯然時(shí)間已經(jīng)過了，不會(huì)執(zhí)行，第二和第三個(gè)則在規(guī)定的時(shí)間點(diǎn)執(zhí)行了，這里還需要注意的是，通過打印可以看出，main函數(shù)執(zhí)行完成后，已經(jīng)開始執(zhí)行main函數(shù)下面的while循環(huán)打印語句了，而在執(zhí)行循環(huán)的過程中，定時(shí)任務(wù)仍然生效，這就是非阻塞式調(diào)度器的原理，如果是阻塞式，則在此代碼中，會(huì)一直卡在main函數(shù)中，main下面額while循環(huán)語句是不會(huì)執(zhí)行的，因此在實(shí)際使用中，非阻塞式應(yīng)用的是非常多的

2022-05-25 14:00:02
2022-05-25 14:00:02
C:\Users\hitre\.virtualenvs\zentaolinkgitlab-QCS5yxD9\lib\site-packages\apscheduler\util.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類型的觸發(fā)器

如下代碼演示了時(shí)間間隔循環(huán)執(zhí)行的使用例子

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秒執(zhí)行一次
? ? sched.add_job(do_func,trigger="interval",args=("張三豐",100),seconds=3)
? ? # 每3分鐘執(zhí)行一次
? ? sched.add_job(do_func, trigger="interval", args=("張三豐", 100), minutes=3)
? ? # 每3小時(shí)執(zhí)行一次
? ? sched.add_job(do_func, trigger="interval", args=("張三豐", 100), hours=3)
? ? # 每3天執(zhí)行一次
? ? sched.add_job(do_func, trigger="interval", args=("張三豐", 100), days=3)
? ? # 每3周執(zhí)行一次
? ? 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)

上面的代碼中因?yàn)闀r(shí)間跨度比較大，這里只演示妹3秒執(zhí)行一次的代碼

代碼如下：

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秒執(zhí)行一次
? ? 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)

執(zhí)行結(jié)果如下：

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類型的觸發(fā)器

cron的觸發(fā)器有點(diǎn)類似linux上crontab定時(shí)器的使用，代碼如下：

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()

? ? # 任務(wù)會(huì)在6月、7月、8月、11月和12月的第三個(gè)周五，00:00、01:00、02:00和03:00觸發(fā)
? ? 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)

這里需要注意的是，可以省略不需要的字段。當(dāng)省略時(shí)間參數(shù)時(shí)，在顯式指定參數(shù)之前的參數(shù)會(huì)被設(shè)定為*，之后的參數(shù)會(huì)被設(shè)定為最小值，week 和day_of_week的最小值為*

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

此外，也可以直接使用crontab表達(dá)式，如下：

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()

? ? # 任務(wù)會(huì)在6月、7月、8月、11月和12月的第三個(gè)周五，00:00、01:00、02:00和03:00觸發(fā)
? ? 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)

四、定時(shí)器使用裝飾器的方法

以間隔時(shí)間循環(huán)執(zhí)行的代碼為例，如下為未使用裝飾器的方式

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秒執(zhí)行一次
? ? 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)

執(zhí)行結(jié)果如下：

2022-05-25 14:34:10
2022-05-25 14:34:11
2022-05-25 14:34:12
2022-05-25 14:34:13 in do func : 姓名：張三豐年齡：100
2022-05-25 14:34:13
2022-05-25 14:34:14
2022-05-25 14:34:15
2022-05-25 14:34:16 in do func : 姓名：張三豐年齡：100
2022-05-25 14:34:16
2022-05-25 14:34:17
2022-05-25 14:34:18
2022-05-25 14:34:19 in do func : 姓名：張三豐年齡：100
2022-05-25 14:34:19
2022-05-25 14:34:20

到此這篇關(guān)于Python使用apscheduler模塊設(shè)置定時(shí)任務(wù)的實(shí)現(xiàn)的文章就介紹到這了,更多相關(guān)Python apscheduler定時(shí)任務(wù)內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: