Python sleep（）：如何向代码中添加时间延迟

您是否需要让Python程序等待一些东西？大多数情况下，您希望代码能够尽快执行。但有时，让代码休眠一段时间实际上符合您的最佳利益。

例如，您可以使用Pythonsleep()调用来模拟程序中的延迟。也许您需要等待文件上传或下载，或者等待图形加载或绘制到屏幕上。您甚至可能需要在调用Web API或查询数据库之间暂停。sleep()在每种情况下，向程序添加Python调用都可以提供帮助，甚至更多！

本文的目标读者是希望增加Python知识的中级开发人员。如果这听起来像你，那我们就开始吧！

使用添加Pythonsleep()调用time.sleep()

>>> import time
>>> time.sleep(3) # Sleep for 3 seconds

$ python3 -m timeit -n 3 "import time; time.sleep(3)"
3 loops, best of 3: 3 sec per loop

import time
import urllib.request
import urllib.error

def uptime_bot(url):
    while True:
        try:
            conn = urllib.request.urlopen(url)
        except urllib.error.HTTPError as e:
            # Email admin / log
            print(f'HTTPError: {e.code} for {url}')
        except urllib.error.URLError as e:
            # Email admin / log
            print(f'URLError: {e.code} for {url}')
        else:
            # Website is up
            print(f'{url} is up')
        time.sleep(60)

if __name__ == '__main__':
    url = 'http://www.google.com/py'
    uptime_bot(url)

HTTPError: 404 for http://www.google.com/py

sleep()使用Decorators添加Python调用

有时需要重试失败的函数。一个流行的使用案例是，当您因为服务器忙而需要重试文件下载时。您通常不想太频繁地向服务器发出请求，因此在每个请求之间添加一个Python调用是可取的。

我亲身经历的另一个用例是在自动测试期间需要检查用户界面的状态。用户界面的加载速度可能比平时快，也可能慢，这取决于我运行测试的计算机。这会改变程序正在验证某个内容时屏幕上的内容。

在这种情况下，我可以让程序休眠片刻，然后在一两秒钟后重新检查。这可能意味着通过测试和失败测试之间的区别。

在这两种情况下，您都可以使用装饰程序添加Pythonsleep()系统调用。如果您不熟悉decorators，或者您想对它们进行深入了解，那么请查看Python decorators入门。让我们看一个例子：

import time
import urllib.request
import urllib.error

def sleep(timeout, retry=3):
    def the_real_decorator(function):
        def wrapper(*args, **kwargs):
            retries = 0
            while retries < retry:
                try:
                    value = function(*args, **kwargs)
                    if value is None:
                        return
                except:
                    print(f'Sleeping for {timeout} seconds')
                    time.sleep(timeout)
                    retries += 1
        return wrapper
    return the_real_decorator

sleep()是您的装饰者。它接受一个timeout值和它应该retry的次数，默认值为3。在sleep()内部是另一个函数，the_real_decorator()，它接受修饰函数。

最后，最内部的函数wrapper()接受传递给修饰函数的参数和关键字参数。这就是魔法发生的地方！使用while循环重试调用函数。如果出现异常，则调用time.sleep()，增加retries计数器，然后再次尝试运行该函数。

现在重写uptime_bot()以使用新的装饰器：

@sleep(3)
def uptime_bot(url):
    try:
        conn = urllib.request.urlopen(url)
    except urllib.error.HTTPError as e:
        # Email admin / log
        print(f'HTTPError: {e.code} for {url}')
        # Re-raise the exception for the decorator
        raise urllib.error.HTTPError
    except urllib.error.URLError as e:
        # Email admin / log
        print(f'URLError: {e.code} for {url}')
        # Re-raise the exception for the decorator
        raise urllib.error.URLError
    else:
        # Website is up
        print(f'{url} is up')

if __name__ == '__main__':
    url = 'http://www.google.com/py'
    uptime_bot(url)

这里，您用3秒的sleep()来装饰uptime_bot()。您还删除了原始的while循环，以及对sleep(60)的旧调用。装饰程序现在处理这个问题。

您所做的另一个更改是在异常处理块中添加一个raise。这是为了装饰工能正常工作。您可以编写decorator来处理这些错误，但是由于这些异常只适用于urllib，因此最好保持decorator的原样。这样，它将适用于更广泛的函数。

注意：如果您想了解Python中的异常处理，请查看Python异常：简介。

您可以对decorator进行一些改进。如果它的重试次数用完，仍然失败，那么您可以让它重新引发最后一个错误。最后一次失败后，装饰程序还会等待3秒钟，这可能是您不希望发生的事情。你可以把这些当作练习来尝试！

sleep()使用线程添加Python调用

有时候，您可能想将Pythonsleep()调用添加到线程中。也许您正在对具有数百万条记录的数据库运行迁移脚本。您不想造成任何停机时间，但是您也不想等待比完成迁移所需的时间更长的时间，因此决定使用线程。

注意：线程是Python中执行并发的一种方法。您可以一次运行多个线程以提高应用程序的吞吐量。如果您不熟悉Python中的线程，请查看Python中的线程简介。

为了防止客户注意到任何类型的减速，每个线程都需要运行一段时间，然后休眠。有两种方法可以做到这一点：

• 像以前一样使用time.sleep()。

• 使用threading模块中的Event.wait()。

让我们先看看time.sleep()

使用time.sleep()

Python日志记录烹饪书展示了一个使用time.sleep()的好例子。Python的logging模块是线程安全的，因此在本练习中，它比print()语句更有用。下面的代码基于此示例：

import logging
import threading
import time

def worker(arg):
    while not arg["stop"]:
        logging.debug("worker thread checking in")
        time.sleep(1)

def main():
    logging.basicConfig(
        level=logging.DEBUG,
        format="%(relativeCreated)6d %(threadName)s %(message)s"
    )
    info = {"stop": False}
    thread = threading.Thread(target=worker, args=(info,))
    thread_two = threading.Thread(target=worker, args=(info,))
    thread.start()
    thread_two.start()

    while True:
        try:
            logging.debug("Checking in from main thread")
            time.sleep(0.75)
        except KeyboardInterrupt:
            info["stop"] = True
            logging.debug('Stopping')
            break
    thread.join()
    thread_two.join()

if __name__ == "__main__":
    main()

这里，您使用Python的threading模块创建两个线程。您还可以创建一个日志对象，将threadName日志记录到stdout。接下来，启动两个线程并每隔一段时间从主线程启动一个循环来记录日志。您可以使用KeyboardInterrupt捕捉用户按下+

尝试在终端中运行上述代码。您将看到类似于以下内容的输出：

0 Thread-1 worker thread checking in
 1 Thread-2 worker thread checking in
 1 MainThread Checking in from main thread
752 MainThread Checking in from main thread
1001 Thread-1 worker thread checking in
1001 Thread-2 worker thread checking in
1502 MainThread Checking in from main thread
2003 Thread-1 worker thread checking in
2003 Thread-2 worker thread checking in
2253 MainThread Checking in from main thread
3005 Thread-1 worker thread checking in
3005 MainThread Checking in from main thread
3005 Thread-2 worker thread checking in

当每个线程运行然后休眠时，日志输出将打印到控制台。现在您已经尝试了一个示例，您将能够在自己的代码中使用这些概念。

使用Event.wait()

模块提供了一个Event()，您可以像time.sleep()一样使用它。然而，Event()还有一个额外的好处，那就是反应更快。原因是当事件被设置时，程序将立即跳出循环。使用time.sleep()时，代码需要等待Pythonsleep()调用完成，线程才能退出。

之所以要在这里使用wait()是因为wait()是非阻塞的，而time.sleep()是阻塞的。这意味着，当您使用time.sleep()时，将阻止主线程在等待sleep()调用结束时继续运行。wait()解决了这个问题。您可以在Python的线程文档中阅读更多关于这些工作原理的信息。

下面是如何使用Event.wait()添加Pythonsleep()调用：

import logging
import threading

def worker(event):
    while not event.isSet():
        logging.debug("worker thread checking in")
        event.wait(1)

def main():
    logging.basicConfig(
        level=logging.DEBUG,
        format="%(relativeCreated)6d %(threadName)s %(message)s"
    )
    event = threading.Event()

    thread = threading.Thread(target=worker, args=(event,))
    thread_two = threading.Thread(target=worker, args=(event,))
    thread.start()
    thread_two.start()

    while not event.isSet():
        try:
            logging.debug("Checking in from main thread")
            event.wait(0.75)
        except KeyboardInterrupt:
            event.set()
            break

if __name__ == "__main__":
    main()

在本例中，您创建threading.Event()并将其传递给worker()。（回想一下，在上一个示例中，您传递了一个dictionary。）接下来，设置循环以检查是否设置了event。如果不是，那么代码将打印一条消息并等待一段时间，然后再次检查。要设置事件，可以按Ctrl+C。设置事件后，worker()将返回，循环将中断，从而结束程序。

注意：如果您想了解有关词典的更多信息，请查看Python中的词典。

请仔细查看上面的代码块。您如何为每个工作线程分配不同的睡眠时间？你能想出来吗？你可以自己解决这个问题！

sleep()使用异步IO添加Python调用

import asyncio

async def main():
    print('Hello ...')
    await asyncio.sleep(1)
    print('... World!')

# Python 3.7+
asyncio.run(main())

import asyncio
import time

async def output(sleep, text):
    await asyncio.sleep(sleep)
    print(text)

async def main():
    print(f"Started: {time.strftime('%X')}")
    await output(1, 'First')
    await output(2, 'Second')
    await output(3, 'Third')
    print(f"Ended: {time.strftime('%X')}")

# Python 3.7+
asyncio.run(main())

import asyncio
import time

async def output(text, sleep):
    while sleep > 0:
        await asyncio.sleep(1)
        print(f'{text} counter: {sleep} seconds')
        sleep -= 1

async def main():
    task_1 = asyncio.create_task(output('First', 1))
    task_2 = asyncio.create_task(output('Second', 2))
    task_3 = asyncio.create_task(output('Third', 3))
    print(f"Started: {time.strftime('%X')}")
    await task_1
    await task_2
    await task_3                                 
    print(f"Ended: {time.strftime('%X')}")

if __name__ == '__main__':
    asyncio.run(main())

import tkinter
import time

class MyApp:
    def __init__(self, parent):
        self.root = parent
        self.root.geometry("400x400")
        self.frame = tkinter.Frame(parent)
        self.frame.pack()
        b = tkinter.Button(text="click me", command=self.delayed)
        b.pack()

    def delayed(self):
        time.sleep(3)

if __name__ == "__main__":
    root = tkinter.Tk()
    app = MyApp(root)
    root.mainloop()

import tkinter

class MyApp:
    def __init__(self, parent):
        self.root = parent
        self.root.geometry("400x400")
        self.frame = tkinter.Frame(parent)
        self.frame.pack()
        self.root.after(3000, self.delayed)

    def delayed(self):
        print('I was delayed')

if __name__ == "__main__":
    root = tkinter.Tk()
    app = MyApp(root)
    root.mainloop()

import wx

class MyFrame(wx.Frame):
    def __init__(self):
        super().__init__(parent=None, title='Hello World')
        wx.CallLater(4000, self.delayed)
        self.Show()

    def delayed(self):
        print('I was delayed')

if __name__ == '__main__':
    app = wx.App()
    frame = MyFrame()
    app.MainLoop()