搜尋
首頁後端開發Python教學Python開發之多個定時任務在單執行緒下執行的實例分析

Python開發之多個定時任務在單執行緒下執行的實例分析

黄舟
黄舟
Jul 27, 2017 pm 04:03 PM
python任務定時

單一執行緒多定時任務 

1、初始版本：

#想法：計時器，說白了就是延時執行指定的程序，目前自己重構python裡面的定時器不太現實，能力達不到，所以延時操作時還得用到系統定時器，不過我們可以改一下規則；把所有要進行定時操作的程序添加到特定列表中，把列表中定時時間最短程序拿出來，進行threading.Timer（time，callback）綁定，等時間超時觸發自定義的callback，執行剛剛列表取出的程序；然後把時間更新，再次把列表中時間最短的程式拿出了，繼續threading.Timer綁定，不斷的迭代循環；當有新的定時任務加入到列表時，把當前的threading.Timer綁定取消，更新清單中的時間，再次取出最短時間，進行threading.Timer綁定......

#程式碼：


import threading
import time

class Timer():
    '''单线程下的定时器'''

    def __init__(self):
        self.queues = []
        self.timer = None
        self.last_time = time.time()

    def start(self):
        item = self.get()
        if item:
            self.timer = threading.Timer(item[0],self.execute)
            self.timer.start()

    def add(self,item):
        print('add',item)
        self.flush_time()
        self.queues.append(item)
        self.queues.sort(key=lambda x:x[0])

        if self.timer:
            self.timer.cancel()
            self.timer = None
        self.start()

    def get(self):
        item = None
        if len(self.queues) > 0:
            item = self.queues[0]
        return item

    def pop(self):
        item = None
        if len(self.queues) > 0:
            item = self.queues.pop(0)
        return item

    def flush_time(self):
        curr_time = time.time()
        for i in self.queues:
            i[0] = i[0] - (curr_time - self.last_time)
        self.last_time = curr_time

    def execute(self):
        # if self.timer:
        #     self.timer.cancel()
        #     self.timer = None
        item = self.pop()
        self.flush_time()
        if item:
            callback = item[1]
            args = item[0]
            callback(args)
        self.start()

執行與輸出：


#
if __name__ == '__main__':    # 检测线程数
    def func():        while True:            print(threading.active_count())
            time.sleep(1)
    
    f1 = threading.Thread(target=func)
    f1.start()    
    import logging
    logging.basicConfig(level=logging.INFO,format="%(asctime)s %(message)s", datefmt="%m/%d/%Y %H:%M:%S [%A]")    def func1(*args):
        logging.info('func1 %s'%args)        # time.sleep(5)
    
    def func2(*args):
        logging.info('func2 %s' % args)        # time.sleep(5)
    def func3(*args):
        logging.info('func3 %s' % args)        # time.sleep(5)
    
    def func4(*args):
        logging.info('func4 %s' % args)        # time.sleep(5)
    
    def func5(*args):
        logging.info('func5 %s' % args)        # time.sleep(5)
    
    
    # 测试
    t1 = Timer()
    logging.info('start')
    t1.add([5,func1])
    time.sleep(0.5)
    t1.add([4,func2])
    time.sleep(0.5)
    t1.add([3,func3])
    time.sleep(0.5)
    t1.add([2,func4])
    time.sleep(0.5)
    t1.add([1,func5])
    time.sleep(5)
    t1.add([1,func1])
    t1.add([2,func2])
    t1.add([3,func3])
    t1.add([4,func4])
    t1.add([5,func5])    
    # 输出
    # 2
    # 07/27/2017 10:36:47 [Thursday] start
    # add [5, <function func1 at 0x000000D79FC77E18>]
    # add [4, <function func2 at 0x000000D79FCA8488>]
    # 3
    # add [3, <function func3 at 0x000000D79FCA8510>]
    # add [2, <function func4 at 0x000000D79FCA8598>]
    # 3
    # add [1, <function func5 at 0x000000D79FCA8620>]
    # 3
    # 07/27/2017 10:36:50 [Thursday] func5 1
    # 07/27/2017 10:36:51 [Thursday] func4 0.498349666595459
    # 3
    # 07/27/2017 10:36:51 [Thursday] func3 0.49782633781433105
    # 07/27/2017 10:36:52 [Thursday] func2 0.49848270416259766
    # 3
    # 07/27/2017 10:36:52 [Thursday] func1 0.48449039459228516
    # 2
    # 2
    # add [1, <function func1 at 0x000000D79FC77E18>]
    # add [2, <function func2 at 0x000000D79FCA8488>]
    # add [3, <function func3 at 0x000000D79FCA8510>]
    # add [4, <function func4 at 0x000000D79FCA8598>]
    # add [5, <function func5 at 0x000000D79FCA8620>]
    # 3
    # 07/27/2017 10:36:55 [Thursday] func1 0.9990766048431396
    # 3
    # 07/27/2017 10:36:56 [Thursday] func2 0.9988017082214355
    # 3
    # 07/27/2017 10:36:57 [Thursday] func3 0.99928879737854
    # 07/27/2017 10:36:58 [Thursday] func4 0.9991350173950195
    # 3
    # 3
    # 07/27/2017 10:36:59 [Thursday] func5 0.9988160133361816

執行程式碼

註：查看程式碼輸出，所有的計時器都按照標定的時間依次執行，非常完美，一切看起來很美好，只是看起來，呵呵噠，當你把func裡面的time.sleep(5)啟用後，線程數蹭蹭的上來了;原因是上個定時器callback還是執行中，下個定時器已經啟動了，這時就又新增了一個線程，哎,失敗

 

2、修訂版本

想法：利用生成者消費者模型,用到threading.Condition條件變數；強制永遠啟用的是一個Timer!

程式碼：


##

import time
import threading
import logging

class NewTimer(threading.Thread):
    &#39;&#39;&#39;单线程下的定时器&#39;&#39;&#39;
    def __init__(self):
        super().__init__()
        self.queues = []
        self.timer = None
        self.cond = threading.Condition()

    def run(self):
        while True:
            # print(&#39;NewTimer&#39;,self.queues)
            self.cond.acquire()
            item = self.get()
            callback = None
            if not item:
                logging.info(&#39;NewTimer wait&#39;)
                self.cond.wait()
            elif item[0] <= time.time():
                new_item = self.pop()
                callback = new_item[1]
            else:
                logging.info(&#39;NewTimer start sys timer and wait&#39;)
                self.timer = threading.Timer(item[0]-time.time(),self.execute)
                self.timer.start()
                self.cond.wait()
            self.cond.release()

            if callback:
                callback(item[0])

    def add(self, item):
        # print(&#39;add&#39;, item)
        self.cond.acquire()
        item[0] = item[0] + time.time()
        self.queues.append(item)
        self.queues.sort(key=lambda x: x[0])
        logging.info(&#39;NewTimer add notify&#39;)
        if self.timer:
            self.timer.cancel()
            self.timer = None
        self.cond.notify()
        self.cond.release()

    def pop(self):
        item = None
        if len(self.queues) > 0:
            item = self.queues.pop(0)
        return item

    def get(self):
        item = None
        if len(self.queues) > 0:
            item = self.queues[0]
        return item

    def execute(self):
        logging.info(&#39;NewTimer execute notify&#39;)
        self.cond.acquire()
        self.cond.notify()
        self.cond.release()

#執行與輸出：


if __name__ == &#39;__main__&#39;:    def func():        while True:            print(threading.active_count())
            time.sleep(1)

    f1 = threading.Thread(target=func)
    f1.start()
    logging.basicConfig(level=logging.INFO,format="%(asctime)s %(message)s", datefmt="%m/%d/%Y %H:%M:%S [%A]")

    newtimer = NewTimer()
    newtimer.start()    def func1(*args):
        logging.info(&#39;func1 %s&#39;%args)
        time.sleep(5)    def func2(*args):
        logging.info(&#39;func2 %s&#39; % args)
        time.sleep(5)    def func3(*args):
        logging.info(&#39;func3 %s&#39; % args)
        time.sleep(5)    def func4(*args):
        logging.info(&#39;func4 %s&#39; % args)
        time.sleep(5)    def func5(*args):
        logging.info(&#39;func5 %s&#39; % args)
        time.sleep(5)

    newtimer.add([5,func1])
    newtimer.add([4,func2])
    newtimer.add([3,func3])
    newtimer.add([2,func4])
    newtimer.add([1,func5])
    time.sleep(1)
    newtimer.add([1,func1])
    newtimer.add([2,func2])
    newtimer.add([3,func3])
    newtimer.add([4,func4])
    newtimer.add([5,func5])# 输出# 2# 07/27/2017 11:26:19 [Thursday] NewTimer wait# 07/27/2017 11:26:19 [Thursday] NewTimer add notify# 07/27/2017 11:26:19 [Thursday] NewTimer add notify# 07/27/2017 11:26:19 [Thursday] NewTimer add notify# 07/27/2017 11:26:19 [Thursday] NewTimer add notify# 07/27/2017 11:26:19 [Thursday] NewTimer add notify# 07/27/2017 11:26:19 [Thursday] NewTimer start sys timer and wait# 07/27/2017 11:26:20 [Thursday] NewTimer execute notify# 4# 07/27/2017 11:26:20 [Thursday] func5 1501125980.2175007# 07/27/2017 11:26:20 [Thursday] NewTimer add notify# 07/27/2017 11:26:20 [Thursday] NewTimer add notify# 07/27/2017 11:26:20 [Thursday] NewTimer add notify# 07/27/2017 11:26:20 [Thursday] NewTimer add notify# 07/27/2017 11:26:20 [Thursday] NewTimer add notify# 3# 3# 3# 3# 3# 07/27/2017 11:26:25 [Thursday] func4 1501125981.2175007# 3# 3# 3# 3# 07/27/2017 11:26:30 [Thursday] func1 1501125981.218279# 3# 3# 3# 3# 3# 3# 07/27/2017 11:26:35 [Thursday] func3 1501125982.2175007# 3# 3# 3# 3# 07/27/2017 11:26:40 [Thursday] func2 1501125982.218279# 3# 3# 3# 3# 3# 07/27/2017 11:26:45 [Thursday] func2 1501125983.2175007# 3# 3# 3# 3# 3# 07/27/2017 11:26:50 [Thursday] func3 1501125983.218279# 3# 3# 3# 3# 3# 07/27/2017 11:26:55 [Thursday] func1 1501125984.2175007# 3# 3# 3# 3# 3# 07/27/2017 11:27:00 [Thursday] func4 1501125984.218279# 3# 3# 3# 3# 3# 07/27/2017 11:27:05 [Thursday] func5 1501125985.218279# 3# 3# 3# 3# 3# 07/27/2017 11:27:10 [Thursday] NewTimer wait

輸出

###rrreee############### ###註：這次無論如何測試線程數也不會蹭蹭的上漲，同時可以實現多定時器任務要求；缺點：用到了兩線程，沒有用到單線程去實現，第二時間精準度問題，需要等待上個定時程式執行完畢，程式才能繼續執行#######

以上是Python開發之多個定時任務在單執行緒下執行的實例分析的詳細內容。更多資訊請關注PHP中文網其他相關文章！

陳述
本文內容由網友自願投稿，版權歸原作者所有。本站不承擔相應的法律責任。如發現涉嫌抄襲或侵權的內容，請聯絡admin@php.cn
相關文章
Python的主要目的：靈活性和易用性Python的主要目的：靈活性和易用性Apr 17, 2025 am 12:14 AM

Python的靈活性體現在多範式支持和動態類型系統，易用性則源於語法簡潔和豐富的標準庫。 1.靈活性：支持面向對象、函數式和過程式編程，動態類型系統提高開發效率。 2.易用性：語法接近自然語言，標準庫涵蓋廣泛功能，簡化開發過程。

Python：多功能編程的力量Python：多功能編程的力量Apr 17, 2025 am 12:09 AM

Python因其簡潔與強大而備受青睞，適用於從初學者到高級開發者的各種需求。其多功能性體現在：1)易學易用，語法簡單；2)豐富的庫和框架，如NumPy、Pandas等；3)跨平台支持，可在多種操作系統上運行；4)適合腳本和自動化任務，提升工作效率。

每天2小時學習Python：實用指南每天2小時學習Python：實用指南Apr 17, 2025 am 12:05 AM

可以，在每天花費兩個小時的時間內學會Python。 1.制定合理的學習計劃，2.選擇合適的學習資源，3.通過實踐鞏固所學知識，這些步驟能幫助你在短時間內掌握Python。

Python與C：開發人員的利弊Python與C：開發人員的利弊Apr 17, 2025 am 12:04 AM

Python適合快速開發和數據處理，而C 適合高性能和底層控制。 1)Python易用，語法簡潔，適用於數據科學和Web開發。 2)C 性能高，控制精確，常用於遊戲和系統編程。

Python：時間投入和學習步伐Python：時間投入和學習步伐Apr 17, 2025 am 12:03 AM

學習Python所需時間因人而異，主要受之前的編程經驗、學習動機、學習資源和方法及學習節奏的影響。設定現實的學習目標並通過實踐項目學習效果最佳。

Python：自動化，腳本和任務管理Python：自動化，腳本和任務管理Apr 16, 2025 am 12:14 AM

Python在自動化、腳本編寫和任務管理中表現出色。 1)自動化：通過標準庫如os、shutil實現文件備份。 2)腳本編寫：使用psutil庫監控系統資源。 3)任務管理：利用schedule庫調度任務。 Python的易用性和豐富庫支持使其在這些領域中成為首選工具。

Python和時間：充分利用您的學習時間Python和時間：充分利用您的學習時間Apr 14, 2025 am 12:02 AM

要在有限的時間內最大化學習Python的效率，可以使用Python的datetime、time和schedule模塊。 1.datetime模塊用於記錄和規劃學習時間。 2.time模塊幫助設置學習和休息時間。 3.schedule模塊自動化安排每週學習任務。

Python：遊戲，Guis等Python：遊戲，Guis等Apr 13, 2025 am 12:14 AM

Python在遊戲和GUI開發中表現出色。 1)遊戲開發使用Pygame，提供繪圖、音頻等功能，適合創建2D遊戲。 2)GUI開發可選擇Tkinter或PyQt，Tkinter簡單易用，PyQt功能豐富，適合專業開發。

See all articles

熱AI工具

Undresser.AI Undress

Undresser.AI Undress

人工智慧驅動的應用程序，用於創建逼真的裸體照片

AI Clothes Remover

AI Clothes Remover

用於從照片中去除衣服的線上人工智慧工具。

Undress AI Tool

Undress AI Tool

免費脫衣圖片

Clothoff.io

Clothoff.io

AI脫衣器

AI Hentai Generator

AI Hentai Generator

免費產生 AI 無盡。

顯示更多

熱門文章

R.E.P.O.能量晶體解釋及其做什麼（黃色晶體）
1 個月前By尊渡假赌尊渡假赌尊渡假赌
R.E.P.O.最佳圖形設置
1 個月前By尊渡假赌尊渡假赌尊渡假赌
刺客信條陰影：貝殼謎語解決方案
2 週前ByDDD
R.E.P.O.如果您聽不到任何人，如何修復音頻
1 個月前By尊渡假赌尊渡假赌尊渡假赌
R.E.P.O.聊天命令以及如何使用它們
1 個月前By尊渡假赌尊渡假赌尊渡假赌
顯示更多

熱工具

SecLists

SecLists

SecLists是最終安全測試人員的伙伴。它是一個包含各種類型清單的集合，這些清單在安全評估過程中經常使用，而且都在一個地方。 SecLists透過方便地提供安全測試人員可能需要的所有列表，幫助提高安全測試的效率和生產力。清單類型包括使用者名稱、密碼、URL、模糊測試有效載荷、敏感資料模式、Web shell等等。測試人員只需將此儲存庫拉到新的測試機上，他就可以存取所需的每種類型的清單。

Safe Exam Browser

Safe Exam Browser

Safe Exam Browser是一個安全的瀏覽器環境，安全地進行線上考試。該軟體將任何電腦變成一個安全的工作站。它控制對任何實用工具的訪問，並防止學生使用未經授權的資源。

Atom編輯器mac版下載

Atom編輯器mac版下載

最受歡迎的的開源編輯器

MinGW - Minimalist GNU for Windows

MinGW - Minimalist GNU for Windows

這個專案正在遷移到osdn.net/projects/mingw的過程中，你可以繼續在那裡關注我們。 MinGW：GNU編譯器集合（GCC）的本機Windows移植版本，可自由分發的導入函式庫和用於建置本機Windows應用程式的頭檔；包括對MSVC執行時間的擴展，以支援C99功能。 MinGW的所有軟體都可以在64位元Windows平台上運作。

Dreamweaver CS6

Dreamweaver CS6

視覺化網頁開發工具

顯示更多

熱門話題

gmail信箱登陸入口在哪裡
7534
15
CakePHP 教程
1379
52
steam的賬戶名稱是什麼格式
82
11
win11激活密鑰永久
55
19
NYT連接提示和答案
21
86
顯示更多