Python-定時器

一. 線程定時器Timer原理

Timer最基本的理解是定時器，可以啟動多個定時任務，這些定時器任務是異步執行，所以不存在等待順序執行順序。

定時器只能執行一次，如果需要重復執行，需要重新添加任務。

導入模塊

from threading import Timer

timer = threading.Timer(interval, function, args=None, kwargs=None)

參數介紹
interval — 定時器間隔，間隔多少秒之后啟動定時器任務(單位：秒)；
function — 線程函數；
args — 線程參數，可以傳遞元組類型數據，默認為空(缺省參數)；
kwargs — 線程參數，可以傳遞字典類型數據，默認為空(缺省參數)；

from threading import Timer
import time

def test1(name):
    print('test1 {}'.format(name))
    #每隔一秒會執行一次
    timer2 = Timer(1, test1, ('bob', ))
    timer2.start()

def test2():
    print('test2')

#前面的5是等待的時間，第三個參數是傳入test1的參數，數組形式
timer = Timer(5, test1, ('alex', ))
timer.start()
# timer.cancel()#取消執行
test2()

執行結果:

 

二. Schedule(調度器)

Schedule是一個第三方輕量級的任務調度模塊，可以按照秒，分，小時，日期或者自定義事件執行時間

方法格式	功能
scheduler.enter(delay, priority, action, argument=(), kwargs={})	在 time 規定的時間后，執行 action 參數指定的函數，其中 argument 和 kwargs 負責為 action 指定的函數傳參，priority 參數執行要執行任務的等級，當同一時間點有多個任務需要執行時，等級越高（ priority 值越小）的任務會優先執行。該函數會返回一個 event，可用來取消該任務。
scheduler.cancel(event)	取消 event 任務。注意，如果 event 參數執行的任務不存在，則會引發 ValueError 錯誤。
scheduler.run(blocking=True)	運行所有需要調度的任務。如果調用該方法的 blocking 參數為 True，該方法將會阻塞線程，直到所有被調度的任務都執行完成。

示例如下:

import threading
import time
from sched import scheduler

def job1(job):
    print('---------------執行任務:{}------------'.format(job))

def job2(job):
    print('---------------執行任務:{}------------'.format(job))

def job3(job):
    print('---------------執行任務:{}------------'.format(job))

def start_job(*jobs):
    #創建任務調度對象
    sche = scheduler()
    # 指定時間后執行job函數
    while True:
        for job in jobs:
            sche.enter(job['interval_time'], i, job['function'], argument=(job['job'],))
        #執行所有調度的任務
        sche.run()

def main():
    while True:
        print('------------------main-------------------------')
        time.sleep(2)

#定義為線程方法傳入的參數
my_list = [{'job': 'python', 'interval_time': 1, 'function': job1},{'job': 'C++', 'interval_time': 5, 'function': job2},{'job': 'Java', 'interval_time': 2, 'function': job3}]
#創建線程
thread = threading.Thread(target = start_job,args =my_list)
#啟動線程
thread.start()
main()

執行結果:

 

三. APScheduler(任務框架)

APScheduler是Python的一個定時任務框架，用於執行周琦或者定時任務

可以基於日期，時間間隔，及類似Linux上的定時任務crontab類型的定時任務

不僅可以添加，刪除定時任務，還可以將任務存儲到數據庫中，實現任務的持久化，使用起來非常方便

 

1. APScheduler組件

(1). Job作業

Job作為APScheduler最小執行單位

創建Job時指定執行的函數，函數中所需參數，Job執行時的一些設置信息

構建說明

id：指定作業的唯一ID

name：指定作業的名字

trigger：apscheduler定義的觸發器，用於確定Job的執行時間，根據設置的trigger規則，計算得到下次執行此job的
時間， 滿足時將會執行

executor：apscheduler定義的執行器，job創建時設置執行器的名字，根據字符串你名字到scheduler獲取到執行此
job的 執行器，執行job指定的函數

max_instances：執行此job的最大實例數，executor執行job時，根據job的id來計算執行次數，根據設置的最大實例數
來確定是否可執行

next_run_time：Job下次的執行時間，創建Job時可以指定一個時間[datetime],不指定的話則默認根據trigger獲取觸
發時間

misfire_grace_time：Job的延遲執行時間，例如Job的計划執行時間是21:00:00，但因服務重啟或其他原因導致
21:00:31才執行，如果設置此key為40,則該job會繼續執行，否則將會丟棄此job

coalesce：Job是否合並執行，是一個bool值。例如scheduler停止20s后重啟啟動，而job的觸發器設置為5s執行
一次，因此此job錯過了4個執行時間，如果設置為是，則會合並到一次執行，否則會逐個執行

func：Job執行的函數

args：Job執行函數需要的位置參數

kwargs：Job執行函數需要的關鍵字參數

 

2. Scheduler工作流程圖

 (1). Scheduler添加job流程

 

 

(2). Scheduler調度流程

 

實例代碼:

import time
import traceback
from apscheduler.schedulers.background import BackgroundScheduler

def python(job):
    print('job {}'.format(job))

def Java(job):
    print('job {}'.format(job))

def C(job):
    print('job {}'.format(job))


class Schedule():
    def __init__(self):
        self.scheduler_jobs = [{'job_id': 'python', 'job_function': python,
                                'job_type': 'interval', 'job_interval_time': 1},
                               {'job_id': 'Java', 'job_function': Java,
                                'job_type': 'interval', 'job_interval_time': 10},
                               {'job_id': 'C', 'job_function': C,
                                'job_type': 'interval', 'job_interval_time': 3}
                               ]
        self.scheduler = BackgroundScheduler()

    def start_schedule(self):
        try:
            for scheduler_job in self.scheduler_jobs:
                if not self.is_exist_scheduler_job(scheduler_job['job_id']):
                    # 執行函數，執行類型，執行id，執行間隔時間，傳入參數
                    print('---------------開始啟動定時任務{}--------------'.format(scheduler_job['job_id']))
                    self.scheduler.add_job(scheduler_job['job_function'], scheduler_job['job_type'],
                                           id=scheduler_job['job_id'], max_instances=1,
                                           seconds=scheduler_job['job_interval_time'], args=(scheduler_job['job_id'], ))
            self.scheduler.start()
        except Exception as e:
            msg = traceback.format_exc()
            print(msg)

    def is_exist_scheduler_job(self, scheduler_job_id):
        ''' 是否存在定時任務 '''
        if self.scheduler.get_job(scheduler_job_id):
            return True
        return False

    def main(self):
        while True:
            print('-----------main------------------')
            time.sleep(2)


if __name__ == '__main__':
    obj = Schedule()
    obj.start_schedule()
    obj.main()

執行結果: