如何每隔x秒重复执行一个函数？

如果漂移不是一个问题

import threading, time

def print_every_n_seconds(n=2):
    while True:
        print(time.ctime())
        time.sleep(n)
    
thread = threading.Thread(target=print_every_n_seconds, daemon=True)
thread.start()

以异步方式输出。

#Tue Oct 16 17:29:40 2018
#Tue Oct 16 17:29:42 2018
#Tue Oct 16 17:29:44 2018

如果运行的任务需要相当长的时间，那么时间间隔将变为2秒+任务时间，因此，如果您需要精确的调度，则不适用于此。

请注意，daemon=True标志意味着该线程不会阻止应用程序关闭。例如，有问题，pytest在运行测试后会无限期地挂起，等待此线程停止。

主要的区别在于，与cron不同的是，异常会永久地终止守护进程。您可能希望使用异常捕获器和日志记录器进行包装。

只需使用

import time

while True:
    print("this will run after every 30 sec")
    #Your code here
    time.sleep(30)

这是从MestreLion的代码进行了改编的版本。 除了原始功能外，此代码还：
1）添加了first_interval，用于在特定时间触发计时器（调用者需要计算并传递first_interval）
2）解决了原始代码中的竞争条件。在原始代码中，如果控制线程无法取消运行中的计时器（“停止计时器并取消计时器操作的执行。这仅在计时器仍处于等待阶段时才起作用。”引自https://docs.python.org/2/library/threading.html），计时器将无限运行。

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

这里有另一种解决方案，不需要使用任何额外的库。

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()

我使用Tkinter的after()方法，它不会"抢占游戏"(就像之前介绍的sched模块一样)，也就是说，它允许其他事情并行运行：

import Tkinter

def do_something1():
  global n1
  n1 += 1
  if n1 == 6: # (Optional condition)
    print "* do_something1() is done *"; return
  # Do your stuff here
  # ...
  print "do_something1() "+str(n1)
  tk.after(1000, do_something1)

def do_something2(): 
  global n2
  n2 += 1
  if n2 == 6: # (Optional condition)
    print "* do_something2() is done *"; return
  # Do your stuff here
  # ...
  print "do_something2() "+str(n2)
  tk.after(500, do_something2)

tk = Tkinter.Tk(); 
n1 = 0; n2 = 0
do_something1()
do_something2()
tk.mainloop()

do_something1() 和 do_something2() 可以并行运行，速度可以任意间隔。这里，第二个函数将以两倍的速度执行。请注意，我已经使用一个简单的计数器作为终止任一函数的条件。您可以使用任何其他条件或者不使用条件使函数运行直到程序终止（例如时钟）。

一个可能的答案：

import time
t=time.time()

while True:
    if time.time()-t>10:
        #run your task here
        t=time.time()

timed-count可以高精度（即<1ms）地执行此操作，因为它与系统时钟同步。它不会随时间漂移，并且不受代码执行时间长短的影响（当然，前提是小于间隔周期）。

以下是一个简单的阻塞式示例：

from timed_count import timed_count

for count in timed_count(60):
    # Execute code here exactly every 60 seconds
    ...

通过在线程中运行，您可以轻松地使其成为非阻塞的：

from threading import Thread
from timed_count import timed_count

def periodic():
    for count in timed_count(60):
        # Execute code here exactly every 60 seconds
        ...

thread = Thread(target=periodic)
thread.start()

显示当前本地时间。

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)

我使用这个来每小时产生60个事件，大多数事件发生在整分钟后的相同秒数：

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) > minutes:
        wait = max(0,(TICK_TIMING-(time.time() % TICK)))
        print ('standard wait: %.2f' % wait)
        time.sleep(wait)
    elif tick_elapsed < 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()

根据实际情况，您可能会得到不同长度的 ticks：

60,60,62,58,60,60,120,30,30,60,60,60,60,60...etc.

但是在60分钟结束时，您将拥有60个刻度；其中大部分将出现在您喜欢的分钟偏移位置。
在我的系统上，我会得到小于1/20秒的典型漂移，直到需要进行校正为止。
这种方法的优点是解决了时钟漂移的问题；如果您正在执行每个tick附加一个项目的操作，并且您期望每小时附加60个项目，则未考虑漂移可能导致二次指标（如移动平均）认为数据太深入过去，从而导致错误的输出。