python多线程定时器资源-CSDN下载

共2个文件

py：2个

21 浏览量 2023-03-11 10:20:27 上传评论收藏 1KB RAR 举报

在Python编程中，多线程和定时器是两个非常重要的概念。多线程允许程序同时执行多个任务，而定时器则能确保程序在特定时间点或以特定间隔执行某些操作。这里我们将深入探讨如何利用Python的`threading`模块来实现多线程定时器。 `threadtimer.py`文件可能包含了一个名为`ThreadTimer`的类，这个类是用来实现定时器功能的。在Python的`threading`模块中，我们可以使用`Thread`类来创建新的线程，而结合`time`模块的`sleep()`函数，我们可以构建一个能在指定时间后启动新线程的定时器。`ThreadTimer`可能有一个构造函数，接受两个参数：一个是线程执行的函数（函数名或可调用对象），另一个是启动该线程的间隔时间。 ```python import threading class ThreadTimer: def __init__(self, function, interval): self.function = function self.interval = interval self.timer = None def start(self): self.timer = threading.Timer(self.interval, self.function) self.timer.start() def cancel(self): if self.timer is not None: self.timer.cancel() ``` 在这个类中，`start`方法会创建一个新的`Timer`实例，它会在`interval`秒后调用`function`。`cancel`方法用于取消已设置的定时器，防止其在未来触发。然后，`main.py`文件可能是主程序，它会使用`ThreadTimer`来定时启动多线程。在`main`函数中，你可以创建多个`ThreadTimer`实例，每个实例对应一个不同的任务（函数）和间隔时间。例如，你可能有一个函数`do_something`，你想让它每分钟执行一次，那么你就可以这样做： ```python def do_something(): print("Doing something...") # 创建一个ThreadTimer实例 timer = ThreadTimer(do_something, 60) timer.start() ``` 这样，`do_something`函数将在每分钟的同一时刻被调用。如果你需要在每次调用之间传递参数，你可以将函数封装成一个接受参数的lambda表达式或者定义一个带参数的辅助函数。在多线程环境中，需要注意线程安全问题。如果多个线程尝试访问和修改共享资源，可能会出现竞态条件。为了防止这类问题，可以使用锁（`threading.Lock`）或其他同步机制来保护关键代码段。此外，Python的全局解释器锁（GIL）限制了在同一时刻只有一个线程执行Python字节码，这可能会影响多线程的性能。在CPU密集型任务中，多线程可能无法充分利用多核处理器的优势。在这种情况下，可以考虑使用多进程（`multiprocessing`模块）代替多线程。总结起来，`python多线程定时器`的实现涉及到Python的`threading`和`time`模块，通过`Thread`和`Timer`类来创建和管理线程，使用定时器控制线程的启动时间。主程序`main.py`使用这些工具来安排多个任务按设定的间隔执行，同时注意线程安全和性能优化。

资源推荐

资源详情

资源评论

收起资源包目录

threadtimer.rar （2个子文件）

threadtimer.py 3KB

main.py 622B

#!/usr/bin/env python # -*- coding: utf-8 -*- import threading import time import queue as Queue class _timerThread(threading.Thread): def __init__(self, t_name,queue,cond,args): threading.Thread.__init__(self, name=t_name) self.threadtimes = [] self.threadFunc = {} self.lasttimes = {} self.queue = queue self.args = args def setNewTimer(self,newobj): if newobj.func == None: self.threadtimes.remove(newobj.secendtime) self.threadFunc[str(newobj.secendtime)] = None self.lasttimes[str(newobj.secendtime)] = None else: if newobj.secendtime in self.threadtimes: self.threadFunc[str(newobj.secendtime)] = newobj.func self.lasttimes[str(newobj.secendtime)] = int(time.time()) else: self.threadtimes.append(newobj.secendtime) self.threadFunc[str(newobj.secendtime)] = newobj.func self.lasttimes[str(newobj.secendtime)] = int(time.time()) def run(self): last_threadtimes_len = len(self.threadtimes) while(True): #print("begin run") if not self.queue.empty(): objtmp = self.queue.get() self.setNewTimer(objtmp) timetmp = int(time.time()) for tx in self.threadtimes: if last_threadtimes_len == 0 and len(self.threadtimes) > 0: self.lasttimes[str(tx)] = timetmp self.threadFunc[str(tx)](self.args) last_threadtimes_len = len(self.threadtimes) if timetmp - self.lasttimes[str(tx)] >= tx and last_threadtimes_len > 0: self.lasttimes[str(tx)] = timetmp self.threadFunc[str(tx)](self.args) self.condition.release() class _timerObj(): def __init__(self,secendt,funct): self.secendtime = secendt self.func = funct class threadtimer(): def __init__(self,args): self.queue = Queue.Queue() self.cond = threading.Condition() self.args = args self.t_thread = _timerThread(str(int(time.time())),self.queue, self.cond,self.args) self._timers = [] self._initTimer() def _initTimer(self): self.t_thread.setDaemon(True) self.t_thread.start() def getTimers(self): return self._timers def setTimer(self,secendTime,Func): objtmp = _timerObj(secendTime,Func) self._timers.append(secendTime) self.queue.put(objtmp) def removeTimer(self,secendTime): objtmp = _timerObj(secendTime,None) self._timers.remove(secendTime) self.queue.put(objtmp)

评论收藏

内容反馈