python 定時任務

轉自:https://www.cnblogs.com/LinTeX9527/p/6181523.html

最近學習到了 python 中兩種開啟定時任務的方法，和大家分享一下心得。

• sched.scheduler()
• threading.Timer()

sched 定時任務

使用sched的套路如下：

s = sched.scheduler(time.time, time.sleep)
s.enter(delay, priority, func1, (arg1, arg2, ...))
s.enter(delay, priority, func2, (arg1, arg2, arg3, ...))
s.run()

第一步新建一個調度器；

第二步添加任務，可以添加多個任務；

第三步讓調度器開始運行。

第二步各參數含義：

• delay　相對於調度器添加這個任務時刻的延時，以秒為單位；
• priority　優先級，數字越小優先級越高；
• func1　任務函數
• (arg1, arg2, ...)　任務函數的參數

import time
import sched

# 第一個工作函數
# 第二個參數 @starttime 表示任務開始的時間
# 很明顯參數在建立這個任務的時候就已經傳遞過去了，至於任務何時開始執行就由調度器決定了
def worker(msg, starttime):
    print u"任務執行的時刻", time.time(), "傳達的消息是", msg, '任務建立時刻', starttime


# 創建一個調度器示例
# 第一參數是獲取時間的函數，第二個參數是延時函數
print('----------  兩個簡單的例子  -------------')
print('程序啟動時刻：', time.time())
s = sched.scheduler(time.time, time.sleep)
s.enter(1, 1, worker, ('hello', time.time()))
s.enter(3, 1, worker, ('world', time.time()))
s.run()  # 這一個 s.run() 啟動上面的兩個任務
print('睡眠２秒前時刻：', time.time())
time.sleep(2)
print('睡眠２秒結束時刻：', time.time())


#　重點關注下面２個任務，建立時間，啟動時間
# ２個任務的建立時間都很好計算，但有沒有發現 "hello world [3]"　的啟動時間比建立時間晚　１３　秒，
# 這不就是２個 sleep　的總延時嗎？所以說啟動時間並不一定就是 delay　能指定的，還需要看具體的程序環境，
# 如果程序堵塞的很厲害，已經浪費了一大段的時間還沒有到 scheduler 能調度這個任務，當 scheduler 能調度這個
# 任務的時候，發現 delay 已經過去了， scheduler 為了彌補“罪過”，會馬上啟動這個任務。

# 任務 "hello world [15]" 就是一個很好的例子，正常情況下，程序沒有阻塞的那么厲害，在scheduler　能調度這個任務的時候
# 發現 delay 還沒到就等待，如果 delay　時間到了就可以在恰好指定的延時調用這個任務。
print('\n\n----------  兩個復雜的例子  -------------')
s.enter(3, 1, worker, ('hello world [3]', time.time()))
print('睡眠７秒前時刻：', time.time())
time.sleep(7)
print('睡眠７秒結束時刻：', time.time())


s.enter(15, 1, worker, ('hello world [15]', time.time()))
print('睡眠６秒前時刻：', time.time())
time.sleep(6)
print('睡眠６秒結束時刻：', time.time())

s.run() # 過了2秒之后，啟動另外一個任務


print('程序結束時刻', time.time())

----------  兩個簡單的例子  -------------
程序啟動時刻： 1481731389.4
任務執行的時刻 1481731390.4 傳達的消息是 hello 任務建立時刻 1481731389.4
任務執行的時刻 1481731392.41 傳達的消息是 world 任務建立時刻 1481731389.4
睡眠２秒前時刻： 1481731392.41
睡眠２秒結束時刻： 1481731394.41


----------  兩個復雜的例子  -------------
睡眠７秒前時刻： 1481731394.41
睡眠７秒結束時刻： 1481731401.42
睡眠６秒前時刻： 1481731401.42
睡眠６秒結束時刻： 1481731407.42
任務執行的時刻 1481731407.42 傳達的消息是 hello world [3] 任務建立時刻 1481731394.41
任務執行的時刻 1481731416.43 傳達的消息是 hello world [15] 任務建立時刻 1481731401.42
程序結束時刻 1481731416.43

自調任務1

任務快結束時利用 scheduler 又重新調用自己讓自己“活過來”。

# 計數器，一個循環任務，總共讓自己執行3次
total = 0 
# 第二個工作函數，自調任務，自己開啟定時並啟動。
def worker2(msg, starttime):
    global total
    total += 1
    print('當前時刻：', time.time(), '消息是：', msg, ' 啟動時間是：', starttime)
    # 只要沒有讓自己調用到第3次，那么繼續重頭開始執行本任務
    if total < 3:
        # 這里的delay 可以重新指定
        s.enter(5, 2, worker2, ('perfect world %d' % (total), time.time()))
        s.run()

print('程序開始時刻：', time.time())
# 開啟自調任務
s.enter(5, 2, worker2, ('perfect world %d' % (total), time.time()))
s.run()
print('程序結束時刻：', time.time())

程序開始時刻： 1481731439.42
當前時刻： 1481731444.43 消息是： perfect world 0  啟動時間是： 1481731439.42
當前時刻： 1481731449.44 消息是： perfect world 1  啟動時間是： 1481731444.43
當前時刻： 1481731454.44 消息是： perfect world 2  啟動時間是： 1481731449.44
程序結束時刻： 1481731454.44

Threading.Timer() 定時任務

from threading import Timer import time def func(msg, starttime): print('程序啟動時刻：', starttime, '當前時刻：', time.time(), '消息內容 --> %s' % (msg)) # 下面的兩個語句和上面的 scheduler 效果一樣的 Timer(5, func, ('hello', time.time())).start() Timer(3, func, ('world', time.time())).start()

程序啟動時刻： 1481731467.28 當前時刻： 1481731470.28 消息內容 --> world
程序啟動時刻： 1481731467.28 當前時刻： 1481731472.28 消息內容 --> hello

循環任務2

利用 threading.Timer() 建立的自調任務

count = 0
def loopfunc(msg,starttime):
    global count
    print('啟動時刻：', starttime, ' 當前時刻：', time.time(), '消息 --> %s' % (msg))
    count += 1
    if count < 3:
        Timer(3, loopfunc, ('world %d' % (count), time.time())).start()

Timer(3, loopfunc, ('world %d' % (count), time.time())).start()

啟動時刻： 1481731476.35  當前時刻： 1481731479.35 消息 --> world 0
啟動時刻： 1481731479.35  當前時刻： 1481731482.35 消息 --> world 1
啟動時刻： 1481731482.35  當前時刻： 1481731485.35 消息 --> world 2

 

自用，測試，待完善......

方法一（推薦使用）

from threading import Timer
import time

def interval_time(timeStr):
    #timeStr = '2018-5-3 09:48:15'  時間格式
    #調用格式 interval_time('2018-5-3 09:48:05')
    timeArray = time.strptime(timeStr, "%Y-%m-%d %H:%M:%S")
    timestamp = time.mktime(timeArray)
    return int(timestamp-time.time())

def func(msg, starttime):
    print('程序啟動時刻：', starttime, '當前時刻：', time.time(), '消息內容 --> %s' % (msg))
    
# 下面的兩個語句和上面的 scheduler 效果一樣的
Timer(interval_time('2018-5-3 09:48:05'), func, ('hello', time.time())).start()
Timer(interval_time('2018-5-3 09:48:15'), func, ('world', time.time())).start()

方法二（使用sched）

#coding=utf-8
#這里需要引入二個模塊，如果需要使用系統cmd，增加os模塊執行cmd命令
import time, sched
import threading  
    
# 第一個參數確定任務的時間，返回從某個特定的時間到現在經歷的秒數  
# 第二個參數以某種人為的方式衡量時間  
schedule = sched.scheduler(time.time, time.sleep)  

# 計算距離當前時間的秒數  
def calc_time(timeStr):
    #timeStr = '2017-3-7 23:59:00'  時間格式
    timeArray = time.strptime(timeStr, "%Y-%m-%d %H:%M:%S")
    timestamp = time.mktime(timeArray)
    return int(timestamp-time.time())

#待運行主函數
def perform_command(id, timeStr):
    print('定時任務 {0} ,計划時間：{1}, 當前時間：'.format(id, timeStr),time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())) 
    # print(arg)
        
#示例函數
def timming_exe(arg, inc = 60):  
    # enter用來安排某事件的發生時間，從現在起第n秒開始啟動
    #enter 延遲時間 優先級 回調函數 參數  
    schedule.enter(inc, 0, perform_command, (arg,))  
    # 持續運行，直到計划時間隊列變成空為止  
    schedule.run() 

#--------以下自編測試，方便調用-----------僅需增加或修改args
#多線程運行主函數
def run_threading(id, timeStr):
    mythread = threading.Thread(target=perform_command, args=(id, timeStr))
    mythread.start()

#寫入計划定時任務的函數
def execute_at_time(id, timeStr, func):
    inc = calc_time(timeStr)
    schedule.enter(inc, 0, func, (id, timeStr))
    #schedule.enter(inc1, 0, func, (id, timeStr))
    #schedule.enter(inc2, 0, func, (id, timeStr))
    #schedule.enter(inc3, 0, func, (id, timeStr))
    schedule.run()

def main():
    execute_at_time('1', '2018-5-2 22:43:20', run_threading)
    execute_at_time('2', '2018-5-2 22:43:30', run_threading)
    execute_at_time('4', '2018-5-2 22:43:40', run_threading)
    execute_at_time('5', '2018-5-2 22:43:50', run_threading)
    execute_at_time('6', '2018-5-2 22:43:55', run_threading)
    execute_at_time('7', '2018-5-2 22:44:20', run_threading)
    execute_at_time('8', '2018-5-2 22:44:30', run_threading)
    execute_at_time('9', '2018-5-2 22:44:50', run_threading)
    execute_at_time('10', '2018-5-2 22:44:55', run_threading)

if __name__ == '__main__':
    main()