Python 中的延迟函数-IT科技

摘要：问题描述：在 JavaScript 中，我习惯于调用稍后执行的函数，如下所示function foo() { alert('bar'); } setTimeout(foo, 1000); 这不会阻止其他代码的执行。我不知道如何在 Python 中实现类似的东西。我可以使用 sleepimport t...

问题描述：

在 JavaScript 中，我习惯于调用稍后执行的函数，如下所示

function foo() {
    alert('bar');
}

setTimeout(foo, 1000);

这不会阻止其他代码的执行。

我不知道如何在 Python 中实现类似的东西。我可以使用 sleep

import time
def foo():
    print('bar')

time.sleep(1)
foo()

但这会阻止其他代码的执行。（实际上，就我而言，阻止 Python 本身不是问题，但我无法对该方法进行单元测试。）

我知道线程是为不同步执行而设计的，但我想知道是否存在更简单、类似setTimeout或setInterval类似的方法。

解决方案 1：

要在延迟后执行某个功能或者使用事件循环（无线程）在给定的秒数内重复某个功能，您可以：

特金特

#!/usr/bin/env python
from Tkinter import Tk

def foo():
    print("timer went off!")

def countdown(n, bps, root):
    if n == 0:
        root.destroy() # exit mainloop
    else:
        print(n)
        root.after(1000 / bps, countdown, n - 1, bps, root)  # repeat the call

root = Tk()
root.withdraw() # don't show the GUI window
root.after(4000, foo) # call foo() in 4 seconds
root.after(0, countdown, 10, 2, root)  # show that we are alive
root.mainloop()
print("done")

输出

10
9
8
7
6
5
4
3
timer went off!
2
1
done

鍵盤

#!/usr/bin/env python
from gi.repository import GObject, Gtk

def foo():
    print("timer went off!")

def countdown(n): # note: a closure could have been used here instead
    if n[0] == 0:
        Gtk.main_quit() # exit mainloop
    else:
        print(n[0])
        n[0] -= 1
        return True # repeat the call

GObject.timeout_add(4000, foo) # call foo() in 4 seconds
GObject.timeout_add(500, countdown, [10])
Gtk.main()
print("done")

输出

10
9
8
7
6
5
4
timer went off!
3
2
1
done

扭曲

#!/usr/bin/env python
from twisted.internet import reactor
from twisted.internet.task import LoopingCall

def foo():
    print("timer went off!")

def countdown(n):
    if n[0] == 0:
        reactor.stop() # exit mainloop
    else:
        print(n[0])
        n[0] -= 1

reactor.callLater(4, foo) # call foo() in 4 seconds
LoopingCall(countdown, [10]).start(.5)  # repeat the call in .5 seconds
reactor.run()
print("done")

输出

10
9
8
7
6
5
4
3
timer went off!
2
1
done

异步

Python 3.4为异步 IO引入了新的临时 API--asyncio模块：

#!/usr/bin/env python3.4
import asyncio

def foo():
    print("timer went off!")

def countdown(n):
    if n[0] == 0:
        loop.stop() # end loop.run_forever()
    else:
        print(n[0])
        n[0] -= 1

def frange(start=0, stop=None, step=1):
    while stop is None or start < stop:
        yield start
        start += step #NOTE: loss of precision over time

def call_every(loop, seconds, func, *args, now=True):
    def repeat(now=True, times=frange(loop.time() + seconds, None, seconds)):
        if now:
            func(*args)
        loop.call_at(next(times), repeat)
    repeat(now=now)

loop = asyncio.get_event_loop()
loop.call_later(4, foo) # call foo() in 4 seconds
call_every(loop, 0.5, countdown, [10]) # repeat the call every .5 seconds
loop.run_forever()
loop.close()
print("done")

输出

10
9
8
7
6
5
4
3
timer went off!
2
1
done

注意：这些方法的界面和行为略有不同。

解决方案 2：

您需要一个Timer来自模块的对象threading。

from threading import Timer
from time import sleep

def foo():
    print "timer went off!"
t = Timer(4, foo)
t.start()
for i in range(11):
    print i
    sleep(.5)

如果您想重复，这里有一个简单的解决方案：不要使用Timer，只需使用Thread但传递一个类似于这样的函数：

def call_delay(delay, repetitions, func, *args, **kwargs):             
    for i in range(repetitions):    
        sleep(delay)
        func(*args, *kwargs)

这不会产生无限循环，因为如果操作不当，这可能会导致线程不会终止，并产生其他令人不快的行为。更复杂的方法可能是使用Event基于的方法，比如这个。

解决方案 3：

像 Javascript 这样的异步回调setTimeout需要事件驱动的架构。

流行的twisted等 Python 异步框架可以满足CallLater您的需求，但这意味着在您的应用程序中采用事件驱动的架构。

另一种选择是使用线程并在线程中休眠。Python 提供了一个计时器来简化等待部分。但是，当您的线程唤醒并执行您的函数时，它将处于单独的线程中，并且必须以线程安全的方式执行它所做的一切。

解决方案 4：

抱歉，我不能发布超过 2 个链接，因此有关更多信息，请查看PEP 380以及最重要的asyncio文档。

除非您坚持使用线程或多处理，否则 asyncio 是解决此类问题的首选方案。它由 GvR 设计和实施，名称为“Tulip”。GvR 在PyCon 2013上推出了它，旨在成为一个事件循环来统治（和标准化）所有事件循环（如 twisted、gevent 等中的事件循环），并使它们彼此兼容。之前已经提到过 asyncio，但 asyncio 的真正威力是通过Yield from释放出来的。

# asyncio is in standard lib for latest python releases (since 3.3)
import asyncio

# there's only one event loop, let's fetch that
loop = asyncio.get_event_loop()

# this is a simple reminder that we're dealing with a coro
@asyncio.coroutine
def f():
    for x in range(10):
        print(x)
        # we return with a coroutine-object from the function, 
        # saving the state of the execution to return to this point later
        # in this case it's a special sleep
        yield from asyncio.sleep(3)

# one of a few ways to insert one-off function calls into the event loop
loop.call_later(10, print, "ding!")
# we insert the above function to run until the coro-object from f is exhausted and 
# raises a StopIteration (which happens when the function would return normally)
# this also stops the loop and cleans up - keep in mind, it's not DEAD but can be restarted
loop.run_until_complete(f())
# this closes the loop - now it's DEAD
loop.close()

================