Python系列之循环定时器

时间:2022-12-12 23:31:04

近期在学习并使用Python开发一些小工具,在这里记录方便回忆,也与各位开始走上这条路的朋友共勉,如有不正确希望指正,谢谢!


开始使用定时器时,度娘了下有没好的例子,本人比较懒,希望能直接使用。确实找到了一些,但是大多只是很直白的代码,自己打算整理一下。

我选用了threading模块中的定时器,使用线程的优势就是可以不干扰现有进程的正常执行。首先我们看下源码:

很简单的封装加上对线程的继承,函数也就是运行和取消,并有案例说明

# The timer class was contributed by Itamar Shtull-Trauring

def Timer(*args, **kwargs):
    """Factory function to create a Timer object.

    Timers call a function after a specified number of seconds:

        t = Timer(30.0, f, args=[], kwargs={})
        t.start()
        t.cancel()     # stop the timer's action if it's still waiting

    """
    return _Timer(*args, **kwargs)

class _Timer(Thread):
    """Call a function after a specified number of seconds:

            t = Timer(30.0, f, args=[], kwargs={})
            t.start()
            t.cancel()     # stop the timer's action if it's still waiting

    """

    def __init__(self, interval, function, args=[], kwargs={}):
        Thread.__init__(self)
        self.interval = interval
        self.function = function
        self.args = args
        self.kwargs = kwargs
        self.finished = Event()

    def cancel(self):
        """Stop the timer if it hasn't finished yet"""
        self.finished.set()

    def run(self):
        self.finished.wait(self.interval)
        if not self.finished.is_set():
            self.function(*self.args, **self.kwargs)
        self.finished.set()

在run函数中 我们不免可以看出,当启动定时器后,一直在等待,然后进行判断set状态,是否执行用户函数,那么所谓的cancel也不是立即取消,而是简单的置状态。判断后也是运行了与cancel一样的代码。所以我们看出这里的定时器是一次性的定时器,而我们需要循环定时器。


现有使用这种定时器进行循环运行的思路是启用2个定时器,进行相互调用。但是是不是逻辑和使用太复杂呢?

那么我们使用更简单的,直接继承timer修改下run函数即可:

class LoopTimer(_Timer):  
    """Call a function after a specified number of seconds:


            t = LoopTimer(30.0, f, args=[], kwargs={})
            t.start()
            t.cancel()     # stop the timer's action if it's still waiting


    """
    def __init__(self, interval, function, args=[], kwargs={}):
        _Timer.__init__(self,interval, function, args, kwargs)


    def run(self):
        '''self.finished.wait(self.interval)
        if not self.finished.is_set():
            self.function(*self.args, **self.kwargs)
        self.finished.set()'''
        while True:
            self.finished.wait(self.interval)
            if self.finished.is_set():
                self.finished.set()
                break
            self.function(*self.args, **self.kwargs)  

那么我们写了简单的例子测试下

def testlooptimer():  
    print("I am loop timer.")     
      
t = LoopTimer(2,testlooptimer)  
t.start()

运行结果:

I am loop timer.
I am loop timer.
I am loop timer.
I am loop timer.