【问题标题】:Run certain code every n seconds [duplicate]每n秒运行一次特定代码[重复]
【发布时间】:2011-03-24 13:03:49
【问题描述】:

有没有办法,例如,每 n 秒打印一次Hello World!? 例如,该程序将遍历我拥有的任何代码,然后在 5 秒后(使用time.sleep())执行该代码。不过,我会使用它来更新文件,而不是打印 Hello World。

例如:

startrepeat("print('Hello World')", .01) # Repeats print('Hello World') ever .01 seconds

for i in range(5):
    print(i)

>> Hello World!
>> 0
>> 1
>> 2
>> Hello World!
>> 3
>> Hello World!
>> 4

【问题讨论】:

标签: python multithreading


【解决方案1】:
import threading

def printit():
  threading.Timer(5.0, printit).start()
  print "Hello, World!"

printit()

# continue with the rest of your code

https://docs.python.org/3/library/threading.html#timer-objects

【讨论】:

  • 好的,修复了,注意你可能还想将Timer创建的线程设置为daemon,以防你想通过完成主线程来干净地中断程序——在在这种情况下,您最好设置t = threading.Timer &c,然后设置t.daemon = True,然后再设置t.start(),就在print "Hello, World!"之前。
  • 这似乎根本无法解决问题......它不会每秒重复。
  • @YanKingYin:尝试运行它。它确实重复。 printit 自行安排。
  • 这最终不会达到递归函数的极限吗?
  • 我该如何阻止它?
【解决方案2】:

我对这个主题的谦虚看法,Alex Martelli 的回答的概括,带有 start() 和 stop() 控制:

from threading import Timer

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.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._timer = Timer(self.interval, self._run)
            self._timer.start()
            self.is_running = True

    def stop(self):
        self._timer.cancel()
        self.is_running = False

用法:

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!

特点:

  • 仅标准库,无外部依赖项
  • 即使计时器已经启动/停止,start() 和 stop() 也可以安全调用多次
  • 要调用的函数可以有位置参数和命名参数
  • 您可以随时更改interval,下次运行后生效。 args、kwargs 甚至 function 也一样!

【讨论】:

  • @JossieCalderon:sleep() 函数位于 Python 标准库的 time 模块中,除了 import 之外不需要任何额外的代码来使用它。但请注意,这是一次阻塞调用,而不是 OP 要求的重复多线程计时器。基本上,sleep() 只是一个暂停,而不是一个计时器。
  • 这很好,但请注意,它不会每 n 秒运行一次作业。相反,它运行一个 n 秒计时器,然后快速启动另一个 n 秒计时器......但是在下一个计时器开始之前有一个延迟。所以这不会每 n 秒精确一次;相反,它会漂移,工作之间的时间超过 n 秒。运行作业以打印当前时间的 1 秒计时器产生:开始... 16:07:42.017682 16:07:43.023215 16:07:44.023626 等。
  • 是的,@eraoul,它确实会漂移,因为没有直接的方法可以使用 Python 的标准库设置真正的 recurring 计时器,所以我“链接”了几个一次性计时器来模拟它。然而,漂移在第一次运行时为 5 毫秒,在下一次运行时为 0.4 毫秒,因此大约需要 200 到 2500 次运行才能漂移一整秒,这对您来说可能或可能不重要。为了在数千次运行中获得更高的准确性,可以根据 RTC 计算下一次运行或使用外部工具,例如 cron
  • 谢谢你。我添加了self._timer.daemon = True 以更好地处理关机并将函数调用移到self._timer.start() 之后立即运行。
  • @Sarthak:停止你说的计时器?您可以向被调用函数添加一些逻辑来检查经过的时间(将当前时间与以前保存的全局开始时间进行比较)或创建另一个 30 分钟计时器,这两种方法都会禁用重复计时器(在第二个计时器的情况下也是如此)
【解决方案3】:
def update():
    import time
    while True:
        print 'Hello World!'
        time.sleep(5)

这将作为一个函数运行。 while True: 使它永远运行。如果需要,您可以随时将其从功能中取出。

【讨论】:

  • 不起作用;它只是永远运行,我不能做任何其他事情。
  • 你还想同时做些什么?
  • 这只是循环运行。您没有在问题中指定您将在此期间做其他事情,所以我认为这就是您需要的。
  • 注意:这不会相当每 5 秒运行一次。它将每5 + N 秒运行一次,其中N 是执行while True: 和time.sleep(5) 之间的时间所花费的时间。虽然与整整 5 秒相比,print 'Hellow World!' 可以忽略不计,但对于其他场景而言,这可能并不重要。我应该注意,使用线程也不会完全达到 5 秒,但它会更接近。
  • 我知道我迟到了 9 年,但这不是一个有效的解决方案,他特别说 ...the program would go through whatever code I had, then once it had been 5 seconds...。
【解决方案4】:

为自己省点精神分裂症,并使用高级 Python 调度程序: http://pythonhosted.org/APScheduler

代码就是这么简单:

from apscheduler.scheduler import Scheduler

sched = Scheduler()
sched.start()

def some_job():
    print "Every 10 seconds"

sched.add_interval_job(some_job, seconds = 10)

....
sched.shutdown()

【讨论】:

  • 首先,子模块被称为“调度器”,带有一个“s”。那里没有课程调度程序。也许是BackgroundScheduler?无论如何,这个答案是不完整的,不起作用。
  • 已经有一段时间了,我想我从网络用户手册中粘贴了代码。上面的代码现在已经更正了(还没有经过测试,但它来自我自己的工作代码,我一直在使用它)。 PS:也许我们正在寻找不同的版本/模块?我确定我的行是“from apscheduler.scheduler import Scheduler”,大写 S 而不是复数。
  • @MadsSkjern:我看到了2.1 branch has apscheduler.scheduler (no s) module。当前分支 3 没有。
  • 现在(2014 年 11 月)查看此答案的任何人都应该知道,这虽然是一个很好的答案,但完全是错误的。上述 cmets 对此进行了讨论。要在当前版本中添加作业,代码将显示为sched.add_job(some_job, 'interval', seconds = 10)。看documentation
【解决方案5】:

这是一个与APScheduler 3.00+兼容的简单示例:

# note that there are many other schedulers available
from apscheduler.schedulers.background import BackgroundScheduler

sched = BackgroundScheduler()

def some_job():
    print('Every 10 seconds')

# seconds can be replaced with minutes, hours, or days
sched.add_job(some_job, 'interval', seconds=10)
sched.start()

...

sched.shutdown()

或者,您可以使用以下内容。与许多替代方案不同,此计时器将精确地每 n 秒执行一次所需的代码(与执行代码所需的时间无关)。因此,如果您无法承受任何漂移,这是一个不错的选择。

import time
from threading import Event, Thread

class RepeatedTimer:

    """Repeat `function` every `interval` seconds."""

    def __init__(self, interval, function, *args, **kwargs):
        self.interval = interval
        self.function = function
        self.args = args
        self.kwargs = kwargs
        self.start = time.time()
        self.event = Event()
        self.thread = Thread(target=self._target)
        self.thread.start()

    def _target(self):
        while not self.event.wait(self._time):
            self.function(*self.args, **self.kwargs)

    @property
    def _time(self):
        return self.interval - ((time.time() - self.start) % self.interval)

    def stop(self):
        self.event.set()
        self.thread.join()


# start timer
timer = RepeatedTimer(10, print, 'Hello world')

# stop timer
timer.stop()

【讨论】:

    【解决方案6】:

    这是一个不会每隔n 秒创建一个新线程的版本:

    from threading import Event, Thread
    
    def call_repeatedly(interval, func, *args):
        stopped = Event()
        def loop():
            while not stopped.wait(interval): # the first call is in `interval` secs
                func(*args)
        Thread(target=loop).start()    
        return stopped.set
    

    该事件用于停止重复:

    cancel_future_calls = call_repeatedly(5, print, "Hello, World")
    # do something else here...
    cancel_future_calls() # stop future calls
    

    见Improve current implementation of a setInterval python

    【讨论】:

    • 值得注意的是,虽然这确实允许在等待 func 再次运行时运行其他东西,但此解决方案并未考虑 func 本身运行所需的时间。所以这个解决方案等待n 秒 between 运行,而不是每隔n 秒运行一次。上述其他解决方案至少更接近于每n 秒运行一次,尽管此解决方案也有其优势。
    • @Mike:是的。如果您点击链接;您会看到明确提到它并建议了相应的解决方案。这取决于具体情况,哪种解决方案更可取。
    • 如果您能解释一下打印函数是否在主线程中被调用,那就太好了。如果是,do something else 代码会在另一个线程中运行吗?
    • @toing_toing 查看代码:“print”函数作为“func”参数传递给“call_repeatedly”。在另一个线程中执行的“循环”中调用“func”(查看Thread(target=loop))。简而言之:不,这里的主线程中没有调用“打印”函数。
    【解决方案7】:

    您可以启动一个单独的线程,其唯一职责是计数 5 秒,更新文件,重复。您不希望这个单独的线程干扰您的主线程。

    【讨论】:

      猜你喜欢
      • 2022-12-11
      • 1970-01-01
      • 2018-12-19
      • 2011-09-18
      • 1970-01-01
      • 1970-01-01
      • 2015-04-28
      • 1970-01-01
      • 1970-01-01
      相关资源
      最近更新 更多