????????如果沒有接觸過定時模塊，現(xiàn)在有個需求讓定時啟動任務(wù)，該怎么實現(xiàn)？我想過幾種方法：

（1）sleep，直接pass，sleep會阻塞，在等待的這段時間里，啥都做不了，效率不太好。

（2）將定時的時間和當前時間想辦法做比對，如果一樣，再調(diào)用要運行的任務(wù)，再配合一個while Ture，應(yīng)該也可以實現(xiàn)吧。

（3）調(diào)用第三方接口，比如我或許可以找到平臺自身的定時模塊，比如windows的鬧鐘、linux的cron，然后用程序去調(diào)用接口。

????????如果不查找資料，讓自己去做定時，估計就會從上面的（2）或者（3）去思考了。幸好python庫里已經(jīng)有這樣一個模塊：apscheduler，可以直接拿來用。省去很多事。

1、模塊功能

時間調(diào)度模塊，可以用于定時、周期性跑任務(wù)。可以添加任務(wù)、刪除任務(wù)。

APScheduler有四大組件：

觸發(fā)器-trigger：觸發(fā)器就是時間到了要調(diào)什么任務(wù)，有三種方式觸發(fā)任務(wù)調(diào)用：date、cron、interval。

• date: 定時，精確到幾年幾月幾日幾點幾分幾秒

• cron: 很靈活，date和interval能表示的，它基本上都可以表示，掌握cron就能應(yīng)對大部分定時場景了。

• interval：周期性執(zhí)行，比如：每隔5秒執(zhí)行一次任務(wù)調(diào)用

任務(wù)存儲器-job store：作業(yè)存儲的位置，默認情況下放在內(nèi)存里，也可以存放在數(shù)據(jù)庫中。

執(zhí)行器-executor：會把任務(wù)放到進程池或者線程池中運行任務(wù)，任務(wù)完成，執(zhí)行器會通知調(diào)度器。默認是線程池。如果你調(diào)用的任務(wù)是計算密集型的，推薦使用進程池。

調(diào)度器-scheduler：我們一般會使用到調(diào)度器，調(diào)度器能干的活很多，比如配置任務(wù)存儲器、配置執(zhí)行器、編輯作業(yè)等等。

2、模塊安裝

pip install apscheduler

3、模塊使用

從我自己使用該模塊的情況，重點應(yīng)該是放在觸發(fā)器上，它是定時模塊的核心。

使用該模塊分三步：定義調(diào)度器、構(gòu)造觸發(fā)器添加任務(wù)、啟動調(diào)度器。

先舉一個最簡單的例子：

from apscheduler.schedulers.blocking import BlockingScheduler
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s-%(levelname)s-%(funcName)s: %(message)s')
logger = logging.getLogger("阿提艾斯")

def func():
 ? ?logger.info("你好，阿提艾斯")
 ? ?

def scheduler_example():
 ? ?# 定義調(diào)度器實例
 ? ?sche = BlockingScheduler(timezone='Asia/Shanghai')
 ? ?# 構(gòu)造觸發(fā)器，添加job
 ? ?sche.add_job(func, 'cron', year="*", month="*", day="*", hour="15", minute="07", second="00")
 ? ?# 啟動調(diào)度器
 ? ?sche.start() ? ?

上面例子的運行效果：

3.1 定義調(diào)度器

調(diào)度器也不止一種，根據(jù)情況選擇合適的調(diào)度器：

BlockingScheduler: 阻塞式調(diào)度器，適用于只跑調(diào)度器的程序

BackgroundScheduler：后臺式調(diào)度器，適用于非阻塞的情況，調(diào)度器會在后臺運行

AsyncIOScheduler： AsyncIO調(diào)度器，還沒用過

TornadoScheduler: Tornado調(diào)度器，還沒用過

TwistedScheduler: Twisted調(diào)度器，還沒用過

QtScheduler: Qt調(diào)度器，還沒用過

假如不想使用調(diào)度器默認的配置，可以參考這個例子：

# 定義調(diào)度器
def get_instance():
 ?executors = {
 ? ?'default': ThreadPoolExecutor(max_workers=30),
 ? ?#'processorpool': ProcessPoolExecutor(max_workers=4)
 ?}
 ?job_defaults = {
 ? ?'coalesce': True, ? ?# 如果系統(tǒng)因為某些原因沒有執(zhí)行任務(wù)，導致任務(wù)累計，為True只運行最后一次，為False則累計的任務(wù)全部跑一遍。
 ? ?'max_instances': 1 ? # 默認情況下，每個作業(yè)只能同時運行一個實例。該參數(shù)可以為調(diào)度程序設(shè)置允許并發(fā)運行的特定作業(yè)的最大實例。
 ?}
 ?sche = BackgroundScheduler(timezone='Asia/Shanghai', executors=executors, job_defaults=job_defaults)
 ?return sche

3.2 構(gòu)造觸發(fā)器添加任務(wù)

這兒是重點！?。?/p>

?add_job方法傳參源碼：

add_job(func, trigger=None, args=None, kwargs=None, id=None, \
 ? ? ? ? ? ?name=None, misfire_grace_time=undefined, coalesce=undefined, \
 ? ? ? ? ? ?max_instances=undefined, next_run_time=undefined, \
 ? ? ? ? ? ?jobstore='default', executor='default', \
 ? ? ? ? ? ?replace_existing=False, **trigger_args)

參數(shù)解釋：

func: 調(diào)用的函數(shù),比如： def hello(name)

trigger: 觸發(fā)器類型，比如：“date”、“cron”、“interval”

args: 傳遞給func的參數(shù)，比如：args=["xxx"]

kwargs: 以字典形式傳遞給func的參數(shù)，比如： kwargs={"name": "atiaisi"}

id: 任務(wù)id，要保證唯一

name: job的名稱，默認是調(diào)用函數(shù)的名字

misfire_grace_time: 在指定的運行時間之后的幾秒鐘，作業(yè)仍然可以允許運行。（處理任務(wù)定時沒啟動的情況）

coalesce: True或False，為True表示只運行任務(wù)累計的最后一次

max_instances: 每個作業(yè)運行實例個數(shù)

next_run_time: datetime類型，不用管觸發(fā)器如何配置，這個字段都會先生效。當然原先配置的觸發(fā)器定時也會生效。

jobstore: 作業(yè)存儲，可以不配置，使用默認

executor: 執(zhí)行器，可以不配置，使用默認

replace_existing: True或False，為True的，job_id相同的話，后面的job會覆蓋之前的job。

各種定時情況如下：

“*”號表示所有

sche = BlockingScheduler(timezone='Asia/Shanghai', executors=executors, job_defaults=job_defaults)

# 每個月的第3個周3的23：59：59執(zhí)行hello
sche.add_job(hello, 'cron', month="*", day=f"{self.num_dict[3]} {self.week_dict[2]}", hour="23", minute="59", second="59")

# 每隔5秒執(zhí)行一次
sche.add_job(hello, 'interval', days=0, hours=0, minutes=0, seconds=5)
sche.add_job(hello, 'interval', seconds=5)

# 排列組合（3月15日6:7:10、3月15日6:7:30、...）這些時間點分別執(zhí)行hello
sche.add_job(hello, 'cron', month='3, 4', day="15， 20", hour="6, 9", minute="7, 8", second="10, 30")

完整例子代碼：https://gitee.com/atiaisi/py_learn/blob/master/apscheduler_test.py

4、遇到的問題

4.1?PytzUsageWarning: The zone attribute is specific to pytz's interface; please migrate to a new time zone provider

解決方法：定義調(diào)度器的時候，把時區(qū)加上，比如：

sche = BlockingScheduler(timezone='Asia/Shanghai')

5、參考資料

5.1 https://apscheduler.readthedocs.io/en/stable/userguide.html

5.2 涉及到的代碼可以在這兒查看：

https://gitee.com/atiaisi/py_learn/blob/master/apscheduler_test.py

日拱一卒無有盡，功不唐捐終入海。