blocks|key|1428302|text|如果您的程序还没有事件循环，那么可以使用sched模块，它实现了一个通用的事件调度程序。|type|unstyled|depth|inlineStyleRanges|entityRanges|offset|length|data|1428303|import+sched,+time
s+=+sched.scheduler(time.time,+time.sleep)
def+do_something(sc):+
++++print("Doing+stuff...")
++++#+do+your+stuff
++++s.enter(60,+1,+do_something,+(sc,))

s.enter(60,+1,+do_something,+(s,))
s.run()|code-block|syntax|javascript|1428304|如果您已经在使用asyncio、trio、tkinter、PyQt5、gobject、kivy等事件循环库，那么只需使用现有事件循环库的方法来调度任务即可。|style|CODE|1428305|entityMap|0|LINK|mutability|MUTABLE|url|http://docs.python.org/library/sched.html^0|K|5|0|0|0|8|7|G|4|L|7|T|5|Z|7|17|4|0^^$0|@$1|2|3|4|5|6|7|U|8|@]|9|@$A|V|B|W|1|X]]|C|$]]|$1|D|3|E|5|F|7|Y|8|@]|9|@]|C|$G|H]]|$1|I|3|J|5|6|7|Z|8|@$A|10|B|11|K|L]|$A|12|B|13|K|L]|$A|14|B|15|K|L]|$A|16|B|17|K|L]|$A|18|B|19|K|L]|$A|1A|B|1B|K|L]]|9|@]|C|$]]|$1|M|3|-4|5|6|7|1C|8|@]|9|@]|C|$]]]|N|$O|$5|P|Q|R|C|$S|T]]]]

If your program doesn't have a event loop already, use the <a href="http://docs.python.org/library/sched.html" rel="noreferrer">sched</a> module, which implements a general purpose event scheduler.

<pre><code>import sched, time
s = sched.scheduler(time.time, time.sleep)
def do_something(sc): 
 print("Doing stuff...")
 # do your stuff
 s.enter(60, 1, do_something, (sc,))

s.enter(60, 1, do_something, (s,))
s.run()
</code></pre>

If you're already using an event loop library like <code>asyncio</code>, <code>trio</code>, <code>tkinter</code>, <code>PyQt5</code>, <code>gobject</code>, <code>kivy</code>, and many others - just schedule the task using your existing event loop library's methods, instead.

blocks|key|1428340|text|这与cron之间的主要区别在于，异常将永远杀死守护进程。您可能希望使用异常捕获器和记录器进行包装。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|1428341|entityMap^0|0^^$0|@$1|2|3|4|5|6|7|D|8|@]|9|@]|A|$]]|$1|B|3|-4|5|6|7|E|8|@]|9|@]|A|$]]]|C|$]]

The main difference between that and cron is that an exception will kill the daemon for good. You might want to wrap with an exception catcher and logger.

blocks|key|1428389|text|您可能需要考虑Twisted，它是一个实现Reactor+Pattern的Python网络库。|type|unstyled|depth|inlineStyleRanges|entityRanges|offset|length|data|1428390|from+twisted.internet+import+task,+reactor

timeout+=+60.0+#+Sixty+seconds

def+doWork():
++++#do+work+here
++++pass

l+=+task.LoopingCall(doWork)
l.start(timeout)+#+call+every+sixty+seconds

reactor.run()|code-block|syntax|javascript|1428391|虽然"while+True:+sleep(60)“可能会起作用，但Twisted可能已经实现了您最终需要的许多特性(守护进程、日志记录或波本斯指出的异常处理)，并且可能是一个更健壮的解决方案|1428392|entityMap|0|LINK|mutability|MUTABLE|url|http://twistedmatrix.com/trac/|1|http://en.wikipedia.org/wiki/Reactor_pattern^0|7|7|0|L|F|1|0|0|0^^$0|@$1|2|3|4|5|6|7|U|8|@]|9|@$A|V|B|W|1|X]|$A|Y|B|Z|1|10]]|C|$]]|$1|D|3|E|5|F|7|11|8|@]|9|@]|C|$G|H]]|$1|I|3|J|5|6|7|12|8|@]|9|@]|C|$]]|$1|K|3|-4|5|6|7|13|8|@]|9|@]|C|$]]]|L|$M|$5|N|O|P|C|$Q|R]]|S|$5|N|O|P|C|$Q|T]]]]

You might want to consider <a href="http://twistedmatrix.com/trac/" rel="noreferrer">Twisted</a> which is a Python networking library that implements the <a href="http://en.wikipedia.org/wiki/Reactor_pattern" rel="noreferrer">Reactor Pattern</a>.

<pre><code>from twisted.internet import task, reactor

timeout = 60.0 # Sixty seconds

def doWork():
 #do work here
 pass

l = task.LoopingCall(doWork)
l.start(timeout) # call every sixty seconds

reactor.run()
</code></pre>

While "while True: sleep(60)" will probably work Twisted probably already implements many of the features that you will eventually need (daemonization, logging or exception handling as pointed out by bobince) and will probably be a more robust solution

blocks|key|1428445|text|我认为更简单的方法是：|type|unstyled|depth|inlineStyleRanges|entityRanges|data|1428446|import+time

def+executeSomething():
++++#code+here
++++time.sleep(60)

while+True:
++++executeSomething()|code-block|syntax|javascript|1428447|这样你的代码就会被执行，然后等待60秒，然后再次执行，等待，执行，等等……不需要把事情复杂化:D|1428448|entityMap^0|0|0|0^^$0|@$1|2|3|4|5|6|7|K|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|L|8|@]|9|@]|A|$E|F]]|$1|G|3|H|5|6|7|M|8|@]|9|@]|A|$]]|$1|I|3|-4|5|6|7|N|8|@]|9|@]|A|$]]]|J|$]]

The easier way I believe to be:

<pre><code>import time

def executeSomething():
 #code here
 time.sleep(60)

while True:
 executeSomething()
</code></pre>

This way your code is executed, then it waits 60 seconds then it executes again, waits, execute, etc...
No need to complicate things :D

blocks|key|1643376|text|不久前我也遇到过类似的问题。也许http://cronus.readthedocs.org能帮上忙？|type|unstyled|depth|inlineStyleRanges|entityRanges|offset|length|data|1643377|对于v0.2，可以使用以下代码片段|1643378|import+cronus.beat+as+beat

beat.set_rate(2)+#+2+Hz
while+beat.true():
++++#+do+some+time+consuming+work+here
++++beat.sleep()+#+total+loop+duration+would+be+0.5+sec|code-block|syntax|javascript|1643379|entityMap|0|LINK|mutability|MUTABLE|url|http://cronus.readthedocs.org/^0|G|T|0|0|0|0^^$0|@$1|2|3|4|5|6|7|S|8|@]|9|@$A|T|B|U|1|V]]|C|$]]|$1|D|3|E|5|6|7|W|8|@]|9|@]|C|$]]|$1|F|3|G|5|H|7|X|8|@]|9|@]|C|$I|J]]|$1|K|3|-4|5|6|7|Y|8|@]|9|@]|C|$]]]|L|$M|$5|N|O|P|C|$Q|R]]]]

I faced a similar problem some time back. May be <a href="http://cronus.readthedocs.org" rel="noreferrer">http://cronus.readthedocs.org</a> might help?

For v0.2, the following snippet works

<pre><code>import cronus.beat as beat

beat.set_rate(2) # 2 Hz
while beat.true():
 # do some time consuming work here
 beat.sleep() # total loop duration would be 0.5 sec
</code></pre>

blocks|key|1643435|text|将您的时间循环锁定到系统时钟，如下所示：|type|unstyled|depth|inlineStyleRanges|entityRanges|data|1643436|import+time
starttime+=+time.time()
while+True:
++++print("tick")
++++time.sleep(60.0+-+((time.time()+-+starttime)+%25+60.0))|code-block|syntax|javascript|1643437|entityMap^0|0|0^^$0|@$1|2|3|4|5|6|7|I|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|J|8|@]|9|@]|A|$E|F]]|$1|G|3|-4|5|6|7|K|8|@]|9|@]|A|$]]]|H|$]]

Lock your time loop to the system clock like this:
<pre><code>import time
starttime = time.time()
while True:
 print(&quot;tick&quot;)
 time.sleep(60.0 - ((time.time() - starttime) % 60.0))
</code></pre>

blocks|key|1643522|text|这是来自MestreLion的代码更新，它避免了随时间的推移而漂移。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|1643523|这里的RepeatedTimer类按照OP的请求，每隔“间隔”秒调用一次给定的函数；调度并不依赖于函数执行所需的时间。我喜欢这个解决方案，因为它没有外部库依赖；这只是一个纯粹的python。|1643524|import+threading+
import+time

class+RepeatedTimer(object):
++def+__init__(self,+interval,+function,+*args,+**kwargs):
++++self._timer+=+None
++++self.interval+=+interval
++++self.function+=+function
++++self.args+=+args
++++self.kwargs+=+kwargs
++++self.is_running+=+False
++++self.next_call+=+time.time()
++++self.start()

++def+_run(self):
++++self.is_running+=+False
++++self.start()
++++self.function(*self.args,+**self.kwargs)

++def+start(self):
++++if+not+self.is_running:
++++++self.next_call+%2B=+self.interval
++++++self._timer+=+threading.Timer(self.next_call+-+time.time(),+self._run)
++++++self._timer.start()
++++++self.is_running+=+True

++def+stop(self):
++++self._timer.cancel()
++++self.is_running+=+False|code-block|syntax|javascript|1643525|示例用法(抄袭自MestreLion的答案)：|1643526|from+time+import+sleep

def+hello(name):
++++print+"Hello+%25s!"+%25+name

print+"starting..."
rt+=+RepeatedTimer(1,+hello,+"World")+#+it+auto-starts,+no+need+of+rt.start()
try:
++++sleep(5)+#+your+long-running+job+goes+here...
finally:
++++rt.stop()+#+better+in+a+try/finally+block+to+make+sure+the+program+ends!|1643527|entityMap^0|0|0|0|0|0^^$0|@$1|2|3|4|5|6|7|O|8|@]|9|@]|A|$]]|$1|B|3|C|5|6|7|P|8|@]|9|@]|A|$]]|$1|D|3|E|5|F|7|Q|8|@]|9|@]|A|$G|H]]|$1|I|3|J|5|6|7|R|8|@]|9|@]|A|$]]|$1|K|3|L|5|F|7|S|8|@]|9|@]|A|$G|H]]|$1|M|3|-4|5|6|7|T|8|@]|9|@]|A|$]]]|N|$]]

Here's an update to the code from MestreLion that avoids drifiting over time. 

The RepeatedTimer class here calls the given function every "interval" seconds as requested by the OP; the schedule doesn't depend on how long the function takes to execute. I like this solution since it doesn't have external library dependencies; this is just pure python.

<pre><code>import threading 
import time

class RepeatedTimer(object):
 def __init__(self, interval, function, *args, **kwargs):
 self._timer = None
 self.interval = interval
 self.function = function
 self.args = args
 self.kwargs = kwargs
 self.is_running = False
 self.next_call = time.time()
 self.start()

 def _run(self):
 self.is_running = False
 self.start()
 self.function(*self.args, **self.kwargs)

 def start(self):
 if not self.is_running:
 self.next_call += self.interval
 self._timer = threading.Timer(self.next_call - time.time(), self._run)
 self._timer.start()
 self.is_running = True

 def stop(self):
 self._timer.cancel()
 self.is_running = False
</code></pre>

Sample usage (copied from MestreLion's answer):

<pre><code>from time import sleep

def hello(name):
 print "Hello %s!" % name

print "starting..."
rt = RepeatedTimer(1, hello, "World") # it auto-starts, no need of rt.start()
try:
 sleep(5) # your long-running job goes here...
finally:
 rt.stop() # better in a try/finally block to make sure the program ends!
</code></pre>

blocks|key|1643598|text|我使用它来导致每小时60个事件，其中大多数事件发生在整分钟后的相同秒数：|type|unstyled|depth|inlineStyleRanges|entityRanges|data|1643599|import+math
import+time
import+random

TICK+=+60+#+one+minute+tick+size
TICK_TIMING+=+59+#+execute+on+59th+second+of+the+tick
TICK_MINIMUM+=+30+#+minimum+catch+up+tick+size+when+lagging

def+set_timing():

++++now+=+time.time()
++++elapsed+=+now+-+info['begin']
++++minutes+=+math.floor(elapsed/TICK)
++++tick_elapsed+=+now+-+info['completion_time']
++++if+(info['tick']%2B1)+>+minutes:
++++++++wait+=+max(0,(TICK_TIMING-(time.time()+%25+TICK)))
++++++++print+('standard+wait:+%25.2f'+%25+wait)
++++++++time.sleep(wait)
++++elif+tick_elapsed+<+TICK_MINIMUM:
++++++++wait+=+TICK_MINIMUM-tick_elapsed
++++++++print+('minimum+wait:+%25.2f'+%25+wait)
++++++++time.sleep(wait)
++++else:
++++++++print+('skip+set_timing();+no+wait')
++++drift+=+((time.time()+-+info['begin'])+-+info['tick']*TICK+-
++++++++TICK_TIMING+%2B+info['begin']%25TICK)
++++print+('drift:+%25.6f'+%25+drift)

info['tick']+=+0
info['begin']+=+time.time()
info['completion_time']+=+info['begin']+-+TICK

while+1:

++++set_timing()

++++print('hello+world')

++++#random+real+world+event
++++time.sleep(random.random()*TICK_MINIMUM)

++++info['tick']+%2B=+1
++++info['completion_time']+=+time.time()|code-block|syntax|javascript|1643600|根据实际情况，您可能会得到长度为的刻度：|1643601|60,60,62,58,60,60,120,30,30,60,60,60,60,60...etc.|1643602|但是在60分钟结束时，您将有60个刻度；并且大多数刻度将出现在您喜欢的分钟的正确偏移量上。|1643603|在我的系统上，我得到了典型的<1/20秒的漂移，直到需要修正为止。|1643604|这种方法的优点是时钟漂移的分辨率；如果你正在做一些事情，比如每个节拍追加一个项目，而你希望每小时追加60个项目，这可能会导致问题。未能考虑漂移可能会导致诸如移动平均值之类的次要指标考虑数据太深，导致错误输出。|1643605|entityMap^0|0|0|0|0|0|0|0^^$0|@$1|2|3|4|5|6|7|S|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|T|8|@]|9|@]|A|$E|F]]|$1|G|3|H|5|6|7|U|8|@]|9|@]|A|$]]|$1|I|3|J|5|D|7|V|8|@]|9|@]|A|$E|F]]|$1|K|3|L|5|6|7|W|8|@]|9|@]|A|$]]|$1|M|3|N|5|6|7|X|8|@]|9|@]|A|$]]|$1|O|3|P|5|6|7|Y|8|@]|9|@]|A|$]]|$1|Q|3|-4|5|6|7|Z|8|@]|9|@]|A|$]]]|R|$]]

I use this to cause 60 events per hour with most events occurring at the same number of seconds after the whole minute:

<pre><code>import math
import time
import random

TICK = 60 # one minute tick size
TICK_TIMING = 59 # execute on 59th second of the tick
TICK_MINIMUM = 30 # minimum catch up tick size when lagging

def set_timing():

 now = time.time()
 elapsed = now - info['begin']
 minutes = math.floor(elapsed/TICK)
 tick_elapsed = now - info['completion_time']
 if (info['tick']+1) &gt; minutes:
 wait = max(0,(TICK_TIMING-(time.time() % TICK)))
 print ('standard wait: %.2f' % wait)
 time.sleep(wait)
 elif tick_elapsed &lt; TICK_MINIMUM:
 wait = TICK_MINIMUM-tick_elapsed
 print ('minimum wait: %.2f' % wait)
 time.sleep(wait)
 else:
 print ('skip set_timing(); no wait')
 drift = ((time.time() - info['begin']) - info['tick']*TICK -
 TICK_TIMING + info['begin']%TICK)
 print ('drift: %.6f' % drift)

info['tick'] = 0
info['begin'] = time.time()
info['completion_time'] = info['begin'] - TICK

while 1:

 set_timing()

 print('hello world')

 #random real world event
 time.sleep(random.random()*TICK_MINIMUM)

 info['tick'] += 1
 info['completion_time'] = time.time()
</code></pre>

Depending upon actual conditions you might get ticks of length:

<pre><code>60,60,62,58,60,60,120,30,30,60,60,60,60,60...etc.
</code></pre>

but at the end of 60 minutes you'll have 60 ticks; and most of them will occur at the correct offset to the minute you prefer.

On my system I get typical drift of &lt; 1/20th of a second until need for correction arises.

The advantage of this method is resolution of clock drift; which can cause issues if you're doing things like appending one item per tick and you expect 60 items appended per hour. Failure to account for drift can cause secondary indications like moving averages to consider data too deep into the past resulting in faulty output.

blocks|key|1643643|text|一个可能的答案是：|type|unstyled|depth|inlineStyleRanges|entityRanges|data|1643644|import+time
t=time.time()

while+True:
++++if+time.time()-t>10:
++++++++#run+your+task+here
++++++++t=time.time()|code-block|syntax|javascript|1643645|entityMap^0|0|0^^$0|@$1|2|3|4|5|6|7|I|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|J|8|@]|9|@]|A|$E|F]]|$1|G|3|-4|5|6|7|K|8|@]|9|@]|A|$]]]|H|$]]

One possible answer:

<pre><code>import time
t=time.time()

while True:
 if time.time()-t&gt;10:
 #run your task here
 t=time.time()
</code></pre>

blocks|key|1428807|text|例如，显示当前本地时间|type|unstyled|depth|inlineStyleRanges|entityRanges|data|1428808|import+datetime
import+glib
import+logger

def+get_local_time():
++++current_time+=+datetime.datetime.now().strftime("%25H:%25M")
++++logger.info("get_local_time():+%25s",current_time)
++++return+str(current_time)

def+display_local_time():
++++logger.info("Current+time+is:+%25s",+get_local_time())
++++return+True

#+call+every+minute
glib.timeout_add(60*1000,+display_local_time)|code-block|syntax|javascript|1428809|entityMap^0|0|0^^$0|@$1|2|3|4|5|6|7|I|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|J|8|@]|9|@]|A|$E|F]]|$1|G|3|-4|5|6|7|K|8|@]|9|@]|A|$]]]|H|$]]

e.g., Display current local time

<pre><code>import datetime
import glib
import logger

def get_local_time():
 current_time = datetime.datetime.now().strftime("%H:%M")
 logger.info("get_local_time(): %s",current_time)
 return str(current_time)

def display_local_time():
 logger.info("Current time is: %s", get_local_time())
 return True

# call every minute
glib.timeout_add(60*1000, display_local_time)
</code></pre>

blocks|key|1428830|text|++++'''+tracking+number+of+times+it+prints'''
import+threading

global+timeInterval
count=0
def+printit():
++threading.Timer(timeInterval,+printit).start()
++print(+"Hello,+World!")
++global+count
++count=count%2B1
++print(count)
printit

if+__name__+==+"__main__":
++++timeInterval=+int(input('Enter+Time+in+Seconds:'))
++++printit()|type|code-block|depth|inlineStyleRanges|entityRanges|data|syntax|javascript|1428831|unstyled|entityMap^0|0^^$0|@$1|2|3|4|5|6|7|G|8|@]|9|@]|A|$B|C]]|$1|D|3|-4|5|E|7|H|8|@]|9|@]|A|$]]]|F|$]]

<pre><code> ''' tracking number of times it prints'''
import threading

global timeInterval
count=0
def printit():
 threading.Timer(timeInterval, printit).start()
 print( "Hello, World!")
 global count
 count=count+1
 print(count)
printit

if __name__ == "__main__":
 timeInterval= int(input('Enter Time in Seconds:'))
 printit()
</code></pre>

blocks|key|1643854|text|以下是来自MestreLion的代码的改编版本。除了原始函数之外，此代码还包括：|type|unstyled|depth|inlineStyleRanges|entityRanges|data|1643855|1)添加用于在特定时间触发定时器的first_interval+(调用方需要计算first_interval并传入)|1643856|2)在原始代码中解决竞争条件。在原始代码中，如果控制线程未能取消正在运行的计时器(“停止计时器，并取消计时器操作的执行。只有当计时器仍处于等待阶段时，这才会起作用。”引用自https://docs.python.org/2/library/threading.html)，计时器将无休止地运行。|offset|length|1643857|class+RepeatedTimer(object):
def+__init__(self,+first_interval,+interval,+func,+*args,+**kwargs):
++++self.timer++++++=+None
++++self.first_interval+=+first_interval
++++self.interval+++=+interval
++++self.func+++=+func
++++self.args+++++++=+args
++++self.kwargs+++++=+kwargs
++++self.running+=+False
++++self.is_started+=+False

def+first_start(self):
++++try:
++++++++#+no+race-condition+here+because+only+control+thread+will+call+this+method
++++++++#+if+already+started+will+not+start+again
++++++++if+not+self.is_started:
++++++++++++self.is_started+=+True
++++++++++++self.timer+=+Timer(self.first_interval,+self.run)
++++++++++++self.running+=+True
++++++++++++self.timer.start()
++++except+Exception+as+e:
++++++++log_print(syslog.LOG_ERR,+"timer+first_start+failed+%25s+%25s"%25(e.message,+traceback.format_exc()))
++++++++raise

def+run(self):
++++#+if+not+stopped+start+again
++++if+self.running:
++++++++self.timer+=+Timer(self.interval,+self.run)
++++++++self.timer.start()
++++self.func(*self.args,+**self.kwargs)

def+stop(self):
++++#+cancel+current+timer+in+case+failed+it's+still+OK
++++#+if+already+stopped+doesn't+matter+to+stop+again
++++if+self.timer:
++++++++self.timer.cancel()
++++self.running+=+False|code-block|syntax|javascript|1643858|entityMap|0|LINK|mutability|MUTABLE|url|https://docs.python.org/2/library/threading.html^0|0|0|2E|1C|0|0|0^^$0|@$1|2|3|4|5|6|7|U|8|@]|9|@]|A|$]]|$1|B|3|C|5|6|7|V|8|@]|9|@]|A|$]]|$1|D|3|E|5|6|7|W|8|@]|9|@$F|X|G|Y|1|Z]]|A|$]]|$1|H|3|I|5|J|7|10|8|@]|9|@]|A|$K|L]]|$1|M|3|-4|5|6|7|11|8|@]|9|@]|A|$]]]|N|$O|$5|P|Q|R|A|$S|T]]]]

Here's an adapted version to the code from MestreLion.
In addition to the original function, this code:

1) add first_interval used to fire the timer at a specific time(caller need to calculate the first_interval and pass in)

2) solve a race-condition in original code. In the original code, if control thread failed to cancel the running timer("Stop the timer, and cancel the execution of the timer’s action. This will only work if the timer is still in its waiting stage." quoted from <a href="https://docs.python.org/2/library/threading.html" rel="nofollow noreferrer">https://docs.python.org/2/library/threading.html</a>), the timer will run endlessly.

<pre><code>class RepeatedTimer(object):
def __init__(self, first_interval, interval, func, *args, **kwargs):
 self.timer = None
 self.first_interval = first_interval
 self.interval = interval
 self.func = func
 self.args = args
 self.kwargs = kwargs
 self.running = False
 self.is_started = False

def first_start(self):
 try:
 # no race-condition here because only control thread will call this method
 # if already started will not start again
 if not self.is_started:
 self.is_started = True
 self.timer = Timer(self.first_interval, self.run)
 self.running = True
 self.timer.start()
 except Exception as e:
 log_print(syslog.LOG_ERR, "timer first_start failed %s %s"%(e.message, traceback.format_exc()))
 raise

def run(self):
 # if not stopped start again
 if self.running:
 self.timer = Timer(self.interval, self.run)
 self.timer.start()
 self.func(*self.args, **self.kwargs)

def stop(self):
 # cancel current timer in case failed it's still OK
 # if already stopped doesn't matter to stop again
 if self.timer:
 self.timer.cancel()
 self.running = False
</code></pre>

blocks|key|1429018|text|我最终使用了schedule模块。API很不错。|type|unstyled|depth|inlineStyleRanges|entityRanges|offset|length|data|1429019|import+schedule
import+time

def+job():
++++print("I'm+working...")

schedule.every(10).minutes.do(job)
schedule.every().hour.do(job)
schedule.every().day.at("10:30").do(job)
schedule.every(5).to(10).minutes.do(job)
schedule.every().monday.do(job)
schedule.every().wednesday.at("13:15").do(job)
schedule.every().minute.at(":17").do(job)

while+True:
++++schedule.run_pending()
++++time.sleep(1)|code-block|syntax|javascript|1429020|entityMap|0|LINK|mutability|MUTABLE|url|https://github.com/dbader/schedule/^0|6|8|0|0|0^^$0|@$1|2|3|4|5|6|7|Q|8|@]|9|@$A|R|B|S|1|T]]|C|$]]|$1|D|3|E|5|F|7|U|8|@]|9|@]|C|$G|H]]|$1|I|3|-4|5|6|7|V|8|@]|9|@]|C|$]]]|J|$K|$5|L|M|N|C|$O|P]]]]

I ended up using the <a href="https://github.com/dbader/schedule/" rel="noreferrer">schedule</a> module. The API is nice.

<pre><code>import schedule
import time

def job():
 print("I'm working...")

schedule.every(10).minutes.do(job)
schedule.every().hour.do(job)
schedule.every().day.at("10:30").do(job)
schedule.every(5).to(10).minutes.do(job)
schedule.every().monday.do(job)
schedule.every().wednesday.at("13:15").do(job)
schedule.every().minute.at(":17").do(job)

while True:
 schedule.run_pending()
 time.sleep(1)
</code></pre>

blocks|key|1643924|text|这是另一个不使用任何额外的库的解决方案。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|1643925|def+delay_until(condition_fn,+interval_in_sec,+timeout_in_sec):
++++"""Delay+using+a+boolean+callable+function.

++++`condition_fn`+is+invoked+every+`interval_in_sec`+until+`timeout_in_sec`.
++++It+can+break+early+if+condition+is+met.

++++Args:
++++++++condition_fn+++++-+a+callable+boolean+function
++++++++interval_in_sec++-+wait+time+between+calling+`condition_fn`
++++++++timeout_in_sec+++-+maximum+time+to+run

++++Returns:+None
++++"""
++++start+=+last_call+=+time.time()
++++while+time.time()+-+start+<+timeout_in_sec:
++++++++if+(time.time()+-+last_call)+>+interval_in_sec:
++++++++++++if+condition_fn()+is+True:
++++++++++++++++break
++++++++++++last_call+=+time.time()|code-block|syntax|javascript|1643926|entityMap^0|0|0^^$0|@$1|2|3|4|5|6|7|I|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|J|8|@]|9|@]|A|$E|F]]|$1|G|3|-4|5|6|7|K|8|@]|9|@]|A|$]]]|H|$]]

Here is another solution without using any extra libaries.

<pre class="lang-py prettyprint-override"><code>def delay_until(condition_fn, interval_in_sec, timeout_in_sec):
 """Delay using a boolean callable function.

 `condition_fn` is invoked every `interval_in_sec` until `timeout_in_sec`.
 It can break early if condition is met.

 Args:
 condition_fn - a callable boolean function
 interval_in_sec - wait time between calling `condition_fn`
 timeout_in_sec - maximum time to run

 Returns: None
 """
 start = last_call = time.time()
 while time.time() - start &lt; timeout_in_sec:
 if (time.time() - last_call) &gt; interval_in_sec:
 if condition_fn() is True:
 break
 last_call = time.time()
</code></pre>

blocks|key|1429155|text|我认为这取决于你想做什么，而你的问题没有具体说明太多细节。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|1429156|对我来说，我想在一个已经多线程的进程中执行一个开销很大的操作。因此，我让leader进程检查时间，只有她执行昂贵的操作(对深度学习模型进行检查点操作)。为了做到这一点，我增加了计数器，以确保已经过去了5、10、15秒，以每5秒保存一次(或者使用带有math.floor的模算术)：|1429157|def+print_every_5_seconds_have_passed_exit_eventually():
++++"""
++++https://stackoverflow.com/questions/3393612/run-certain-code-every-n-seconds
++++https://stackoverflow.com/questions/474528/what-is-the-best-way-to-repeatedly-execute-a-function-every-x-seconds
++++:return:
++++"""
++++opts+=+argparse.Namespace(start=time.time())
++++next_time_to_print+=+0
++++while+True:
++++++++current_time_passed+=+time.time()+-+opts.start
++++++++if+current_time_passed+>=+next_time_to_print:
++++++++++++next_time_to_print+%2B=+5
++++++++++++print(f'worked+and+{current_time_passed=}')
++++++++++++print(f'{current_time_passed+%25+5=}')
++++++++++++print(f'{math.floor(current_time_passed+%25+5)+==+0}')|code-block|syntax|javascript|1429158|starting+__main__+at+__init__
worked+and+current_time_passed=0.0001709461212158203
current_time_passed+%25+5=0.0001709461212158203
True
worked+and+current_time_passed=5.0
current_time_passed+%25+5=0.0
True
worked+and+current_time_passed=10.0
current_time_passed+%25+5=0.0
True
worked+and+current_time_passed=15.0
current_time_passed+%25+5=0.0
True|1429159|对我来说，检查if语句就是我所需要的。在我已经复杂的多处理多gpu代码中拥有线程和调度器并不是我想要增加的复杂性，如果我可以避免的话，看起来我可以做到。检查worker+id很容易确保只有一个进程在执行此操作。|1429160|请注意，我使用True+print语句来确保模块化算术技巧有效，因为检查确切的时间显然不会起作用！但令我惊喜的是，地板做到了这一点。|1429161|entityMap^0|0|0|0|0|0|0^^$0|@$1|2|3|4|5|6|7|Q|8|@]|9|@]|A|$]]|$1|B|3|C|5|6|7|R|8|@]|9|@]|A|$]]|$1|D|3|E|5|F|7|S|8|@]|9|@]|A|$G|H]]|$1|I|3|J|5|F|7|T|8|@]|9|@]|A|$G|H]]|$1|K|3|L|5|6|7|U|8|@]|9|@]|A|$]]|$1|M|3|N|5|6|7|V|8|@]|9|@]|A|$]]|$1|O|3|-4|5|6|7|W|8|@]|9|@]|A|$]]]|P|$]]

I think it depends what you want to do and your question didn't specify lots of details.
For me I want to do an expensive operation in one of my already multithreaded processes. So I have that leader process check the time and only her do the expensive op (checkpointing a deep learning model). To do this I increase the counter to make sure 5 then 10 then 15 seconds have passed to save every 5 seconds (or use modular arithmetic with math.floor):
<pre><code>def print_every_5_seconds_have_passed_exit_eventually():
 &quot;&quot;&quot;
 https://stackoverflow.com/questions/3393612/run-certain-code-every-n-seconds
 https://stackoverflow.com/questions/474528/what-is-the-best-way-to-repeatedly-execute-a-function-every-x-seconds
 :return:
 &quot;&quot;&quot;
 opts = argparse.Namespace(start=time.time())
 next_time_to_print = 0
 while True:
 current_time_passed = time.time() - opts.start
 if current_time_passed &gt;= next_time_to_print:
 next_time_to_print += 5
 print(f'worked and {current_time_passed=}')
 print(f'{current_time_passed % 5=}')
 print(f'{math.floor(current_time_passed % 5) == 0}')
</code></pre>
<pre><code>starting __main__ at __init__
worked and current_time_passed=0.0001709461212158203
current_time_passed % 5=0.0001709461212158203
True
worked and current_time_passed=5.0
current_time_passed % 5=0.0
True
worked and current_time_passed=10.0
current_time_passed % 5=0.0
True
worked and current_time_passed=15.0
current_time_passed % 5=0.0
True
</code></pre>
To me the check of the if statement is what I need. Having threads, schedulers in my already complicated multiprocessing multi-gpu code is not a complexity I want to add if I can avoid it and it seems I can. Checking the worker id is easy to make sure only 1 process is doing this.
Note I used the True print statements to really make sure the modular arithemtic trick worked since checking for exact time is obviously not going to work! But to my pleasant surprised the floor did the trick.

I want to repeatedly execute a function in Python every 60 seconds forever (just like an <a href="http://web.archive.org/web/20090823012700/http://developer.apple.com:80/DOCUMENTATION/Cocoa/Reference/Foundation/Classes/NSTimer_Class/Reference/NSTimer.html" rel="noreferrer">NSTimer</a> in Objective C or setTimeout in JS). This code will run as a daemon and is effectively like calling the python script every minute using a cron, but without requiring that to be set up by the user.
In <a href="https://stackoverflow.com/questions/373335/suggestions-for-a-cron-like-scheduler-in-python">this question about a cron implemented in Python</a>, the solution appears to effectively just <a href="http://docs.python.org/library/time.html#time.sleep" rel="noreferrer">sleep()</a> for x seconds. I don't need such advanced functionality so perhaps something like this would work
<pre><code>while True:
 # Code executed here
 time.sleep(60)
</code></pre>
Are there any foreseeable problems with this code?

What is the best way to repeatedly execute a function every x seconds?

深度学习

Python

我希望永远每隔60秒重复执行一次Python语言中的函数(就像Objective C中的或JS中的setTimeout一样)。这段代码将作为守护进程运行，实际上就像使用cron每分钟调用一次python脚本，但不需要用户进行设置。在中，解决方案似乎只需 x秒即可。我不需要这么高级的功能，所以这样的功能可能会起作用。wh...

问每隔x秒重复执行一个函数的最佳方式是什么？
EN

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问每隔x秒重复执行一个函数的最佳方式是什么？EN

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问每隔x秒重复执行一个函数的最佳方式是什么？
EN