Python 是否有类似于 JavaScript 的功能setInterval()？
我想拥有：
def set_interval(func, interval):
...
这将调用func每个interval时间单位。
问问题
72330 次
22 回答
52
这可能是您正在寻找的正确片段：
import threading
def set_interval(func, sec):
def func_wrapper():
set_interval(func, sec)
func()
t = threading.Timer(sec, func_wrapper)
t.start()
return t
于 2012-12-26T01:59:27.830 回答
28
这是您可以开始和停止的版本。它没有阻塞。由于未添加执行时间错误，因此也没有故障（例如，对于以非常短的间隔作为音频的长时间执行很重要）
import time, threading
StartTime=time.time()
def action() :
print('action ! -> time : {:.1f}s'.format(time.time()-StartTime))
class setInterval :
def __init__(self,interval,action) :
self.interval=interval
self.action=action
self.stopEvent=threading.Event()
thread=threading.Thread(target=self.__setInterval)
thread.start()
def __setInterval(self) :
nextTime=time.time()+self.interval
while not self.stopEvent.wait(nextTime-time.time()) :
nextTime+=self.interval
self.action()
def cancel(self) :
self.stopEvent.set()
# start action every 0.6s
inter=setInterval(0.6,action)
print('just after setInterval -> time : {:.1f}s'.format(time.time()-StartTime))
# will stop interval in 5s
t=threading.Timer(5,inter.cancel)
t.start()
输出是：
just after setInterval -> time : 0.0s
action ! -> time : 0.6s
action ! -> time : 1.2s
action ! -> time : 1.8s
action ! -> time : 2.4s
action ! -> time : 3.0s
action ! -> time : 3.6s
action ! -> time : 4.2s
action ! -> time : 4.8s
于 2018-02-09T15:44:38.207 回答
20
保持简单。
import threading
def setInterval(func,time):
e = threading.Event()
while not e.wait(time):
func()
def foo():
print "hello"
# using
setInterval(foo,5)
# output:
hello
hello
.
.
.
编辑：此代码是非阻塞的
import threading
class ThreadJob(threading.Thread):
def __init__(self,callback,event,interval):
'''runs the callback function after interval seconds
:param callback: callback function to invoke
:param event: external event for controlling the update operation
:param interval: time in seconds after which are required to fire the callback
:type callback: function
:type interval: int
'''
self.callback = callback
self.event = event
self.interval = interval
super(ThreadJob,self).__init__()
def run(self):
while not self.event.wait(self.interval):
self.callback()
event = threading.Event()
def foo():
print "hello"
k = ThreadJob(foo,event,2)
k.start()
print "It is non-blocking"
于 2015-06-27T12:38:00.883 回答
11
稍微改变一下 Nailxx的答案，你就得到了答案！
from threading import Timer
def hello():
print "hello, world"
Timer(30.0, hello).start()
Timer(30.0, hello).start() # after 30 seconds, "hello, world" will be printed
于 2013-11-03T11:49:24.870 回答
7
该sched模块为通用 Python 代码提供了这些功能。threading.Timer但是，正如其文档所暗示的，如果您的代码是多线程的，则改用该类可能更有意义。
于 2010-04-23T08:03:42.947 回答
6
我想这就是你所追求的：
#timertest.py
import sched, time
def dostuff():
print "stuff is being done!"
s.enter(3, 1, dostuff, ())
s = sched.scheduler(time.time, time.sleep)
s.enter(3, 1, dostuff, ())
s.run()
如果您在重复方法的末尾向调度程序添加另一个条目，它将继续进行。
于 2012-01-04T17:32:00.617 回答
4
import functools
import sched, time
s = sched.scheduler(time.time, time.sleep)
def setInterval(sec):
def decorator(func):
@functools.wraps(func)
def wrapper(*argv, **kw):
setInterval(sec)(func)
func(*argv, **kw)
s.enter(sec, 1, wrapper, ())
return wrapper
s.run()
return decorator
@setInterval(sec=3)
def testInterval():
print ("test Interval ")
testInterval()
于 2020-05-26T19:28:08.947 回答
3
简单的 setInterval 工具
from threading import Timer
def setInterval(timer, task):
isStop = task()
if not isStop:
Timer(timer, setInterval, [timer, task]).start()
def hello():
print "do something"
return False # return True if you want to stop
if __name__ == "__main__":
setInterval(2.0, hello) # every 2 seconds, "do something" will be printed
于 2019-01-07T07:33:55.780 回答
2
上面的解决方案如果出现程序关闭的情况，并不能保证它会优雅的关闭，总是建议通过软杀来关闭程序，大部分都没有停止的功能我找到了一篇不错的文章在 Sankalp 编写的媒体上解决了这两个问题（在 python 中运行定期任务），请参阅附加链接以获得更深入的了解。在下面的示例中，一个名为 signal 的库用于跟踪杀死是软杀死还是硬杀死
import threading, time, signal
from datetime import timedelta
WAIT_TIME_SECONDS = 1
class ProgramKilled(Exception):
pass
def foo():
print time.ctime()
def signal_handler(signum, frame):
raise ProgramKilled
class Job(threading.Thread):
def __init__(self, interval, execute, *args, **kwargs):
threading.Thread.__init__(self)
self.daemon = False
self.stopped = threading.Event()
self.interval = interval
self.execute = execute
self.args = args
self.kwargs = kwargs
def stop(self):
self.stopped.set()
self.join()
def run(self):
while not self.stopped.wait(self.interval.total_seconds()):
self.execute(*self.args, **self.kwargs)
if __name__ == "__main__":
signal.signal(signal.SIGTERM, signal_handler)
signal.signal(signal.SIGINT, signal_handler)
job = Job(interval=timedelta(seconds=WAIT_TIME_SECONDS), execute=foo)
job.start()
while True:
try:
time.sleep(1)
except ProgramKilled:
print "Program killed: running cleanup code"
job.stop()
break
#output
#Tue Oct 16 17:47:51 2018
#Tue Oct 16 17:47:52 2018
#Tue Oct 16 17:47:53 2018
#^CProgram killed: running cleanup code
于 2019-12-10T14:46:19.437 回答
1
上面的大多数答案都没有正确关闭线程。在使用 Jupyter notebook 时，我注意到当发送显式中断时，线程仍在运行，更糟糕的是，它们会从 1 个线程运行、2、4 等开始不断增加。我下面的方法基于@doom 的答案，但很干净通过在主线程中运行无限循环以侦听 SIGINT 和 SIGTERM 事件来处理中断
- 无漂移
- 可取消
- 很好地处理 SIGINT 和 SIGTERM
- 不会为每次运行创建一个新线程
随时提出改进建议
import time
import threading
import signal
# Record the time for the purposes of demonstration
start_time=time.time()
class ProgramKilled(Exception):
"""
An instance of this custom exception class will be thrown everytime we get an SIGTERM or SIGINT
"""
pass
# Raise the custom exception whenever SIGINT or SIGTERM is triggered
def signal_handler(signum, frame):
raise ProgramKilled
# This function serves as the callback triggered on every run of our IntervalThread
def action() :
print('action ! -> time : {:.1f}s'.format(time.time()-start_time))
# https://stackoverflow.com/questions/2697039/python-equivalent-of-setinterval
class IntervalThread(threading.Thread) :
def __init__(self,interval,action, *args, **kwargs) :
super(IntervalThread, self).__init__()
self.interval=interval
self.action=action
self.stopEvent=threading.Event()
self.start()
def run(self) :
nextTime=time.time()+self.interval
while not self.stopEvent.wait(nextTime-time.time()) :
nextTime+=self.interval
self.action()
def cancel(self) :
self.stopEvent.set()
def main():
# Handle SIGINT and SIFTERM with the help of the callback function
signal.signal(signal.SIGTERM, signal_handler)
signal.signal(signal.SIGINT, signal_handler)
# start action every 1s
inter=IntervalThread(1,action)
print('just after setInterval -> time : {:.1f}s'.format(time.time()-start_time))
# will stop interval in 500s
t=threading.Timer(500,inter.cancel)
t.start()
# https://www.g-loaded.eu/2016/11/24/how-to-terminate-running-python-threads-using-signals/
while True:
try:
time.sleep(1)
except ProgramKilled:
print("Program killed: running cleanup code")
inter.cancel()
break
if __name__ == "__main__":
main()
于 2019-02-26T11:49:43.697 回答
1
上述方法对我来说并没有完全做到，因为我需要能够取消间隔。我把这个函数变成了一个类，并想出了以下内容：
class setInterval():
def __init__(self, func, sec):
def func_wrapper():
self.t = threading.Timer(sec, func_wrapper)
self.t.start()
func()
self.t = threading.Timer(sec, func_wrapper)
self.t.start()
def cancel(self):
self.t.cancel()
于 2016-10-04T00:00:46.187 回答
1
你也可以试试这个方法：
import time
while True:
time.sleep(5)
print("5 seconds has passed")
所以它会每 5 秒打印一次“5 seconds has passed”。
该函数sleep()在给定的秒数内暂停执行。该参数可以是一个浮点数，以指示更精确的睡眠时间。
于 2021-10-13T07:04:46.607 回答
0
最近，我和你有同样的问题。我找到了这些解决方案：
1.你可以使用库：threading.Time（上面有介绍）
2.你可以使用库：sched（上面也有介绍）
3.可以使用库：Advanced Python Scheduler（推荐）
于 2014-03-18T01:19:58.593 回答
0
我的 Python 3 模块jsinterval.py会很有帮助！这里是：
"""
Threaded intervals and timeouts from JavaScript
"""
import threading, sys
__all__ = ['TIMEOUTS', 'INTERVALS', 'setInterval', 'clearInterval', 'setTimeout', 'clearTimeout']
TIMEOUTS = {}
INTERVALS = {}
last_timeout_id = 0
last_interval_id = 0
class Timeout:
"""Class for all timeouts."""
def __init__(self, func, timeout):
global last_timeout_id
last_timeout_id += 1
self.timeout_id = last_timeout_id
TIMEOUTS[str(self.timeout_id)] = self
self.func = func
self.timeout = timeout
self.threadname = 'Timeout #%s' %self.timeout_id
def run(self):
func = self.func
delx = self.__del__
def func_wrapper():
func()
delx()
self.t = threading.Timer(self.timeout/1000, func_wrapper)
self.t.name = self.threadname
self.t.start()
def __repr__(self):
return '<JS Timeout set for %s seconds, launching function %s on timeout reached>' %(self.timeout, repr(self.func))
def __del__(self):
self.t.cancel()
class Interval:
"""Class for all intervals."""
def __init__(self, func, interval):
global last_interval_id
self.interval_id = last_interval_id
INTERVALS[str(self.interval_id)] = self
last_interval_id += 1
self.func = func
self.interval = interval
self.threadname = 'Interval #%s' %self.interval_id
def run(self):
func = self.func
interval = self.interval
def func_wrapper():
timeout = Timeout(func_wrapper, interval)
self.timeout = timeout
timeout.run()
func()
self.t = threading.Timer(self.interval/1000, func_wrapper)
self.t.name = self.threadname
self.t.run()
def __repr__(self):
return '<JS Interval, repeating function %s with interval %s>' %(repr(self.func), self.interval)
def __del__(self):
self.timeout.__del__()
def setInterval(func, interval):
"""
Create a JS Interval: func is the function to repeat, interval is the interval (in ms)
of executing the function.
"""
temp = Interval(func, interval)
temp.run()
idx = int(temp.interval_id)
del temp
return idx
def clearInterval(interval_id):
try:
INTERVALS[str(interval_id)].__del__()
del INTERVALS[str(interval_id)]
except KeyError:
sys.stderr.write('No such interval "Interval #%s"\n' %interval_id)
def setTimeout(func, timeout):
"""
Create a JS Timeout: func is the function to timeout, timeout is the timeout (in ms)
of executing the function.
"""
temp = Timeout(func, timeout)
temp.run()
idx = int(temp.timeout_id)
del temp
return idx
def clearTimeout(timeout_id):
try:
TIMEOUTS[str(timeout_id)].__del__()
del TIMEOUTS[str(timeout_id)]
except KeyError:
sys.stderr.write('No such timeout "Timeout #%s"\n' %timeout_id)
代码编辑：
修复了内存泄漏（由@benjaminz 发现）。现在所有线程都在结束时被清理。为什么会发生这种泄漏？它的发生是因为隐式（甚至显式）引用。在我的情况下，TIMEOUTS和INTERVALS。超时自动自我清理（在此补丁之后），因为它们使用函数包装器调用该函数然后自我终止。但这是怎么发生的？gc除非也删除所有引用或使用模块，否则无法从内存中删除对象。解释：没有办法（在我的代码中）创建对超时/间隔的不需要的引用。他们只有一个推荐人：TIMEOUTS/INTERVALS听写。并且，当中断或完成时（只有超时可以不间断地完成），它们会删除对自己的唯一现有引用：它们对应的 dict 元素。使用 完美封装类__all__，因此没有内存泄漏空间。
于 2017-07-16T10:27:16.547 回答
0
我已经编写了代码来在 python 中创建一个非常灵活的 setInterval。给你：
import threading
class AlreadyRunning(Exception):
pass
class IntervalNotValid(Exception):
pass
class setInterval():
def __init__(this, func=None, sec=None, args=[]):
this.running = False
this.func = func # the function to be run
this.sec = sec # interval in second
this.Return = None # The returned data
this.args = args
this.runOnce = None # asociated with run_once() method
this.runOnceArgs = None # asociated with run_once() method
if (func is not None and sec is not None):
this.running = True
if (not callable(func)):
raise TypeError("non-callable object is given")
if (not isinstance(sec, int) and not isinstance(sec, float)):
raise TypeError("A non-numeric object is given")
this.TIMER = threading.Timer(this.sec, this.loop)
this.TIMER.start()
def start(this):
if (not this.running):
if (not this.isValid()):
raise IntervalNotValid("The function and/or the " +
"interval hasn't provided or invalid.")
this.running = True
this.TIMER = threading.Timer(this.sec, this.loop)
this.TIMER.start()
else:
raise AlreadyRunning("Tried to run an already run interval")
def stop(this):
this.running = False
def isValid(this):
if (not callable(this.func)):
return False
cond1 = not isinstance(this.sec, int)
cond2 = not isinstance(this.sec, float)
if (cond1 and cond2):
return False
return True
def loop(this):
if (this.running):
this.TIMER = threading.Timer(this.sec, this.loop)
this.TIMER.start()
function_, Args_ = this.func, this.args
if (this.runOnce is not None): # someone has provide the run_once
runOnce, this.runOnce = this.runOnce, None
result = runOnce(*(this.runOnceArgs))
this.runOnceArgs = None
# if and only if the result is False. not accept "None"
# nor zero.
if (result is False):
return # cancel the interval right now
this.Return = function_(*Args_)
def change_interval(this, sec):
cond1 = not isinstance(sec, int)
cond2 = not isinstance(sec, float)
if (cond1 and cond2):
raise TypeError("A non-numeric object is given")
# prevent error when providing interval to a blueprint
if (this.running):
this.TIMER.cancel()
this.sec = sec
# prevent error when providing interval to a blueprint
# if the function hasn't provided yet
if (this.running):
this.TIMER = threading.Timer(this.sec, this.loop)
this.TIMER.start()
def change_next_interval(this, sec):
if (not isinstance(sec, int) and not isinstance(sec, float)):
raise TypeError("A non-numeric object is given")
this.sec = sec
def change_func(this, func, args=[]):
if (not callable(func)):
raise TypeError("non-callable object is given")
this.func = func
this.args = args
def run_once(this, func, args=[]):
this.runOnce = func
this.runOnceArgs = args
def get_return(this):
return this.Return
您可以获得许多功能和灵活性。运行此代码不会冻结您的代码，您可以在运行时更改间隔，可以在运行时更改函数，可以传递参数，可以从函数中获取返回的对象等等。你也可以耍花招！
这是一个非常简单和基本的使用示例：
import time
def interval(name="world"):
print(f"Hello {name}!")
# function named interval will be called every two seconds
# output: "Hello world!"
interval1 = setInterval(interval, 2)
# function named interval will be called every 1.5 seconds
# output: "Hello Jane!"
interval2 = setInterval(interval, 1.5, ["Jane"])
time.sleep(5) #stop all intervals after 5 seconds
interval1.stop()
interval2.stop()
查看我的 Github 项目以查看更多示例并关注后续更新：D https://github.com/Hzzkygcs/setInterval-python
于 2020-03-25T15:50:23.143 回答
0
这是一个低时间漂移解决方案，它使用线程定期向 Event 对象发出信号。线程的 run() 在等待超时时几乎什么都不做；因此低时间漂移。
# Example of low drift (time) periodic execution of a function.
import threading
import time
# Thread that sets 'flag' after 'timeout'
class timerThread (threading.Thread):
def __init__(self , timeout , flag):
threading.Thread.__init__(self)
self.timeout = timeout
self.stopFlag = False
self.event = threading.Event()
self.flag = flag
# Low drift run(); there is only the 'if'
# and 'set' methods between waits.
def run(self):
while not self.event.wait(self.timeout):
if self.stopFlag:
break
self.flag.set()
def stop(self):
stopFlag = True
self.event.set()
# Data.
printCnt = 0
# Flag to print.
printFlag = threading.Event()
# Create and start the timer thread.
printThread = timerThread(3 , printFlag)
printThread.start()
# Loop to wait for flag and print time.
while True:
global printCnt
# Wait for flag.
printFlag.wait()
# Flag must be manually cleared.
printFlag.clear()
print(time.time())
printCnt += 1
if printCnt == 3:
break;
# Stop the thread and exit.
printThread.stop()
printThread.join()
print('Done')
于 2017-11-20T19:13:23.507 回答
0
上面的一些答案使用func_wrapper并且threading.Timer确实有效，除了它每次调用间隔时都会产生一个新线程，这会导致内存问题。
下面的基本示例通过将间隔放在单独的线程上大致实现了类似的机制。它以给定的时间间隔休眠。在开始编写代码之前，您需要了解以下一些限制：
JavaScript 是单线程的，所以当里面的函数
setInterval被触发时，其他任何东西都不会同时工作（不包括工作线程，但让我们谈谈 . 的一般用例setInterval。因此，线程是安全的。但是在这个实现中，你可能会遇到竞争条件，除非使用threading.rLock.下面的实现
time.sleep是用来模拟区间的，但是加上 的执行时间func，这个区间的总时间可能会比你预期的要大。因此，根据用例，您可能希望“少睡”（减去通话时间func）我只是粗略地测试了这个，你绝对不应该像我那样使用全局变量，随意调整它以适应你的系统。
废话不多说，代码如下：
# Python 2.7
import threading
import time
class Interval(object):
def __init__(self):
self.daemon_alive = True
self.thread = None # keep a reference to the thread so that we can "join"
def ticktock(self, interval, func):
while self.daemon_alive:
time.sleep(interval)
func()
num = 0
def print_num():
global num
num += 1
print 'num + 1 = ', num
def print_negative_num():
global num
print '-num = ', num * -1
intervals = {} # keep track of intervals
g_id_counter = 0 # roughly generate ids for intervals
def set_interval(interval, func):
global g_id_counter
interval_obj = Interval()
# Put this interval on a new thread
t = threading.Thread(target=interval_obj.ticktock, args=(interval, func))
t.setDaemon(True)
interval_obj.thread = t
t.start()
# Register this interval so that we can clear it later
# using roughly generated id
interval_id = g_id_counter
g_id_counter += 1
intervals[interval_id] = interval_obj
# return interval id like it does in JavaScript
return interval_id
def clear_interval(interval_id):
# terminate this interval's while loop
intervals[interval_id].daemon_alive = False
# kill the thread
intervals[interval_id].thread.join()
# pop out the interval from registry for reusing
intervals.pop(interval_id)
if __name__ == '__main__':
num_interval = set_interval(1, print_num)
neg_interval = set_interval(3, print_negative_num)
time.sleep(10) # Sleep 10 seconds on main thread to let interval run
clear_interval(num_interval)
clear_interval(neg_interval)
print "- Are intervals all cleared?"
time.sleep(3) # check if both intervals are stopped (not printing)
print "- Yup, time to get beers"
预期输出：
num + 1 = 1
num + 1 = 2
-num = -2
num + 1 = 3
num + 1 = 4
num + 1 = 5
-num = -5
num + 1 = 6
num + 1 = 7
num + 1 = 8
-num = -8
num + 1 = 9
num + 1 = 10
-num = -10
Are intervals all cleared?
Yup, time to get beers
于 2017-06-23T17:11:46.707 回答
0
睡着直到下一个seconds长度间隔开始：（非并发）
def sleep_until_next_interval(self, seconds):
now = time.time()
fall_asleep = seconds - now % seconds
time.sleep(fall_asleep)
while True:
sleep_until_next_interval(10) # 10 seconds - worktime
# work here
简单，没有漂移。
于 2020-04-15T12:51:35.647 回答
0
setInterval 应该在多个线程上运行，并且在循环运行时不要冻结任务。
这是我的支持多线程功能的 RUNTIME 包：
- setTimeout(F,ms) : 在独立线程中触发函数的时间。
- delayF(F,ms) ：类似 setTimeout(F,ms)。
- setInterval(F,ms) : 异步循环 .pause, .resume : 暂停和恢复间隔
- clearInterval(interval) : 清除间隔
它简短而简单。注意python如果直接输入函数需要lambda，但是lambda不支持命令块，所以需要先定义函数内容，再放入setInterval。
### DEMO PYTHON MULTITHREAD ASYNCHRONOUS LOOP ###
import time;
import threading;
import random;
def delay(ms):time.sleep(ms/1000); # Controil while speed
def setTimeout(R,delayMS):
t=threading.Timer(delayMS/1000,R)
t.start();
return t;
def delayF(R,delayMS):
t=threading.Timer(delayMS/1000,R)
t.start();
return t;
class THREAD:
def __init__(this):
this.R_onRun=None;
this.thread=None;
def run(this):
this.thread=threading.Thread(target=this.R_onRun);
this.thread.start();
def isRun(this): return this.thread.isAlive();
class setInterval :
def __init__(this,R_onRun,msInterval) :
this.ms=msInterval;
this.R_onRun=R_onRun;
this.kStop=False;
this.thread=THREAD();
this.thread.R_onRun=this.Clock;
this.thread.run();
def Clock(this) :
while not this.kStop :
this.R_onRun();
delay(this.ms);
def pause(this) :
this.kStop=True;
def stop(this) :
this.kStop=True;
def resume(this) :
if (this.kStop) :
this.kStop=False;
this.thread.run();
def clearInterval(Timer): Timer.stop();
# EXAMPLE
def p():print(random.random());
tm=setInterval(p,20);
tm2=setInterval(lambda:print("AAAAA"),20);
delayF(tm.pause,1000);
delayF(tm.resume,2000);
delayF(lambda:clearInterval(tm),3000);
保存到文件 .py 并运行它。您将看到它同时打印随机数和字符串“AAAAA”。打印编号线程将在 1 秒后暂停打印并再次恢复打印 1 秒然后停止，而打印字符串保持打印文本不损坏。
如果您使用 OpenCV进行图形动画并使用那些 setInterval 来提高动画速度，则必须有 1 个主线程来应用 waitKey，否则无论延迟多慢或在子线程中应用 waitKey，窗口都会冻结：
def p:... # Your drawing task
setInterval(p,1); # Subthread1 running draw
setInterval(p,1); # Subthread2 running draw
setInterval(p,1); # Subthread3 running draw
while True: cv2.waitKey(10); # Main thread which waitKey have effect
于 2021-12-22T04:17:09.410 回答
-1
Python 中的工作方式有所不同：您需要sleep()（如果要阻止当前线程）或启动一个新线程。请参阅http://docs.python.org/library/threading.html
于 2010-04-23T08:04:03.437 回答
-1
这是一些简单的事情：
import time
delay = 10 # Seconds
def setInterval():
print('I print in intervals!')
time.sleep(delay)
setInterval()
于 2020-11-09T10:30:26.630 回答
-3
来自Python 文档：
from threading import Timer
def hello():
print "hello, world"
t = Timer(30.0, hello)
t.start() # after 30 seconds, "hello, world" will be printed
于 2010-04-23T08:07:11.300 回答