您是否需要讓Python程序等待一些東西？大多數(shù)情況下，您希望代碼能夠盡快執(zhí)行。但有時(shí)，讓代碼休眠一段時(shí)間實(shí)際上符合您的最佳利益。

例如，您可以使用Pythonsleep()調(diào)用來模擬程序中的延遲。也許您需要等待文件上傳或下載，或者等待圖形加載或繪制到屏幕上。您甚至可能需要在調(diào)用Web API或查詢數(shù)據(jù)庫之間暫停。sleep()在每種情況下，向程序添加Python調(diào)用都可以提供幫助，甚至更多！

本文的目標(biāo)讀者是希望增加Python知識(shí)的中級(jí)開發(fā)人員。如果這聽起來像你，那我們就開始吧！

使用添加Pythonsleep()調(diào)用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調(diào)用

有時(shí)需要重試失敗的函數(shù)。一個(gè)流行的使用案例是，當(dāng)您因?yàn)榉?wù)器忙而需要重試文件下載時(shí)。您通常不想太頻繁地向服務(wù)器發(fā)出請(qǐng)求，因此在每個(gè)請(qǐng)求之間添加一個(gè)Python調(diào)用是可取的。

我親身經(jīng)歷的另一個(gè)用例是在自動(dòng)測試期間需要檢查用戶界面的狀態(tài)。用戶界面的加載速度可能比平時(shí)快，也可能慢，這取決于我運(yùn)行測試的計(jì)算機(jī)。這會(huì)改變程序正在驗(yàn)證某個(gè)內(nèi)容時(shí)屏幕上的內(nèi)容。

在這種情況下，我可以讓程序休眠片刻，然后在一兩秒鐘后重新檢查。這可能意味著通過測試和失敗測試之間的區(qū)別。

在這兩種情況下，您都可以使用裝飾程序添加Pythonsleep()系統(tǒng)調(diào)用。如果您不熟悉decorators，或者您想對(duì)它們進(jìn)行深入了解，那么請(qǐng)查看Python decorators入門。讓我們看一個(gè)例子：

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()是您的裝飾者。它接受一個(gè)timeout值和它應(yīng)該retry的次數(shù)，默認(rèn)值為3。在sleep()內(nèi)部是另一個(gè)函數(shù)，the_real_decorator()，它接受修飾函數(shù)。

最后，最內(nèi)部的函數(shù)wrapper()接受傳遞給修飾函數(shù)的參數(shù)和關(guān)鍵字參數(shù)。這就是魔法發(fā)生的地方！使用while循環(huán)重試調(diào)用函數(shù)。如果出現(xiàn)異常，則調(diào)用time.sleep()，增加retries計(jì)數(shù)器，然后再次嘗試運(yùn)行該函數(shù)。

現(xiàn)在重寫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循環(huán)，以及對(duì)sleep(60)的舊調(diào)用。裝飾程序現(xiàn)在處理這個(gè)問題。

您所做的另一個(gè)更改是在異常處理塊中添加一個(gè)raise。這是為了裝飾工能正常工作。您可以編寫decorator來處理這些錯(cuò)誤，但是由于這些異常只適用于urllib，因此最好保持decorator的原樣。這樣，它將適用于更廣泛的函數(shù)。

注意：如果您想了解Python中的異常處理，請(qǐng)查看Python異常：簡介。

您可以對(duì)decorator進(jìn)行一些改進(jìn)。如果它的重試次數(shù)用完，仍然失敗，那么您可以讓它重新引發(fā)最后一個(gè)錯(cuò)誤。最后一次失敗后，裝飾程序還會(huì)等待3秒鐘，這可能是您不希望發(fā)生的事情。你可以把這些當(dāng)作練習(xí)來嘗試！

sleep()使用線程添加Python調(diào)用

有時(shí)候，您可能想將Pythonsleep()調(diào)用添加到線程中。也許您正在對(duì)具有數(shù)百萬條記錄的數(shù)據(jù)庫運(yùn)行遷移腳本。您不想造成任何停機(jī)時(shí)間，但是您也不想等待比完成遷移所需的時(shí)間更長的時(shí)間，因此決定使用線程。

注意：線程是Python中執(zhí)行并發(fā)的一種方法。您可以一次運(yùn)行多個(gè)線程以提高應(yīng)用程序的吞吐量。如果您不熟悉Python中的線程，請(qǐng)查看Python中的線程簡介。

為了防止客戶注意到任何類型的減速，每個(gè)線程都需要運(yùn)行一段時(shí)間，然后休眠。有兩種方法可以做到這一點(diǎn)：

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

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

讓我們先看看time.sleep()

使用time.sleep()

Python日志記錄烹飪書展示了一個(gè)使用time.sleep()的好例子。Python的logging模塊是線程安全的，因此在本練習(xí)中，它比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模塊創(chuàng)建兩個(gè)線程。您還可以創(chuàng)建一個(gè)日志對(duì)象，將threadName日志記錄到stdout。接下來，啟動(dòng)兩個(gè)線程并每隔一段時(shí)間從主線程啟動(dòng)一個(gè)循環(huán)來記錄日志。您可以使用KeyboardInterrupt捕捉用戶按下+

嘗試在終端中運(yùn)行上述代碼。您將看到類似于以下內(nèi)容的輸出：

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

當(dāng)每個(gè)線程運(yùn)行然后休眠時(shí)，日志輸出將打印到控制臺(tái)。現(xiàn)在您已經(jīng)嘗試了一個(gè)示例，您將能夠在自己的代碼中使用這些概念。

使用Event.wait()

模塊提供了一個(gè)Event()，您可以像time.sleep()一樣使用它。然而，Event()還有一個(gè)額外的好處，那就是反應(yīng)更快。原因是當(dāng)事件被設(shè)置時(shí)，程序?qū)⒘⒓刺鲅h(huán)。使用time.sleep()時(shí)，代碼需要等待Pythonsleep()調(diào)用完成，線程才能退出。

之所以要在這里使用wait()是因?yàn)?code>wait()是非阻塞的，而time.sleep()是阻塞的。這意味著，當(dāng)您使用time.sleep()時(shí)，將阻止主線程在等待sleep()調(diào)用結(jié)束時(shí)繼續(xù)運(yùn)行。wait()解決了這個(gè)問題。您可以在Python的線程文檔中閱讀更多關(guān)于這些工作原理的信息。

下面是如何使用Event.wait()添加Pythonsleep()調(diào)用：

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

在本例中，您創(chuàng)建threading.Event()并將其傳遞給worker()。（回想一下，在上一個(gè)示例中，您傳遞了一個(gè)dictionary。）接下來，設(shè)置循環(huán)以檢查是否設(shè)置了event。如果不是，那么代碼將打印一條消息并等待一段時(shí)間，然后再次檢查。要設(shè)置事件，可以按Ctrl+C。設(shè)置事件后，worker()將返回，循環(huán)將中斷，從而結(jié)束程序。

注意：如果您想了解有關(guān)詞典的更多信息，請(qǐng)查看Python中的詞典。

請(qǐng)仔細(xì)查看上面的代碼塊。您如何為每個(gè)工作線程分配不同的睡眠時(shí)間？你能想出來嗎？你可以自己解決這個(gè)問題！

sleep()使用異步IO添加Python調(diào)用

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