Python sleep():如何向代码中添加时间延迟

您所在的位置:网站首页 延迟函数代码 Python sleep():如何向代码中添加时间延迟

Python sleep():如何向代码中添加时间延迟

2023-03-13 12:07| 来源: 网络整理| 查看: 265

Python sleep():如何向代码中添加时间延迟 python sleep() 时间延迟 2021-02-03 11:45:20 834 0 0

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

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

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

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

内置支持使程序进入睡眠状态。time模块有一个函数sleep(),您可以使用它来暂停调用线程的执行,不管您指定了多少秒。

下面是一个如何使用time.sleep()的示例:

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

如果您在控制台中运行此代码,那么您应该会遇到一个延迟可以在REPL中输入新语句。

注意:在Python3.5中,核心开发人员稍微更改了time.sleep()的行为。新的Pythonsleep()系统调用将至少持续指定的秒数,即使睡眠被信号中断。但是,如果信号本身引发异常,则这不适用。

您可以使用Python的timeit模块测试睡眠持续时间:

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

在这里,您使用-n参数运行timeit模块,该参数告诉timeit运行语句的次数接下来就是。您可以看到timeit运行语句3次,最佳运行时间是3秒,这是预期的。

默认的timeit运行代码的次数是一百万次。如果您使用默认的-n运行上述代码,那么每次迭代3秒时,您的终端将挂起大约34天!timeit模块还有其他几个命令行选项,您可以在它的文档中查看这些选项。

让我们创建一些更真实的选项。系统管理员需要知道他们的网站何时关闭。您希望能够定期检查网站的状态代码,但不能经常查询web服务器,否则会影响性能。执行此检查的一种方法是使用Pythonsleep()系统调用:

import timeimport urllib.requestimport 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)

在这里创建uptime_bot(),它将URL作为其参数。然后,该函数尝试使用urllib打开该URL。如果有一个HTTPError或URLError,那么程序将捕获它并打印出错误。(在实时环境中,您将记录错误,并可能向网站管理员或系统管理员发送电子邮件。)

如果没有出现错误,那么您的代码将打印出一切正常。不管发生什么,你的程序都会休眠60秒。这意味着你每分钟只访问一次网站。本例中使用的URL不正确,因此它将每分钟向控制台输出一次以下内容:

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

继续并更新代码以使用已知的好URL,如http://www.google.com。然后您可以重新运行它以查看它是否成功工作。您还可以尝试更新代码以发送电子邮件或记录错误。有关如何执行此操作的详细信息,请查看使用Python发送电子邮件和登录Python。

sleep()使用Decorators添加Python调用

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

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

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

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

import timeimport urllib.requestimport 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 loggingimport threadingimport 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 thread752 MainThread Checking in from main thread1001 Thread-1 worker thread checking in1001 Thread-2 worker thread checking in1502 MainThread Checking in from main thread2003 Thread-1 worker thread checking in2003 Thread-2 worker thread checking in2253 MainThread Checking in from main thread3005 Thread-1 worker thread checking in3005 MainThread Checking in from main thread3005 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 loggingimport 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调用

异步功能已在3.4版本中添加到Python,此功能集从那时以来一直在积极扩展。异步编程是一种并行编程,它允许您一次运行多个任务。任务完成后,它将通知主线程。

asyncio是一个允许您异步添加Python调用的模块。如果您不熟悉Python的异步编程实现,请查看Python中的异步IO:完整的演练和Python并发与并行编程。

这里有一个来自Python自己文档的示例:

import asyncio

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

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

在这个示例中,您运行main()并让它休眠两次调用之间的一秒钟。

以下是asyncio文档中的协同程序和任务部分中一个更引人注目的示例:

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

在这段代码中,您创建了一个名为output()的工作进程,它占用到sleep和text的秒数打印出来。然后,使用Python的await关键字等待output()代码运行。这里需要await,因为output()已标记为async函数,所以不能像调用普通函数那样调用它。

运行此代码时,程序将执行await3次。代码将等待1、2和3秒,总等待时间为6秒。您还可以重写代码,使任务并行运行:

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

现在您使用的是任务的概念,您可以使用create_task()来创建它。在asyncio中使用任务时,Python将异步运行这些任务。因此,当您运行上述代码时,它应该在3秒内完成,而不是6秒。

sleep()使用GUI添加Python调用

命令行应用程序并不是唯一需要添加Pythonsleep()调用的地方。创建图形用户界面(GUI)时,有时需要添加延迟。例如,您可以创建一个FTP应用程序来下载数百万个文件,但您需要在批处理之间添加一个调用,这样就不会使服务器陷入困境。

GUI代码将在一个名为事件循环的主线程中运行其所有处理和绘图。如果在GUI代码中使用time.sleep(),则会阻塞其事件循环。从用户的角度来看,应用程序可能会冻结。当应用程序使用此方法休眠时,用户将无法与应用程序交互。(在Windows上,您甚至可能会收到有关应用程序现在如何无响应的警报。)

幸运的是,除了time.sleep()之外,您还可以使用其他方法。在接下来的几节中,您将学习如何在Tkinter和wxPython中添加Python调用。如果您在Linux或Mac上使用的是Python的预装版本,那么它可能不可用。如果您得到一个ImportError,那么您需要研究如何将它添加到您的系统中。但是如果您自己安装Python,那么应该已经可以使用了。运行此代码以查看添加Python调用时发生的情况:

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

运行代码后,按GUI中的按钮。当按钮等待sleep()完成时,它将向下停留3秒钟。如果应用程序有其他按钮,那么您将无法单击它们。您也不能在应用程序睡眠时关闭它,因为它无法响应关闭事件。

要使tkinter正常睡眠,您需要使用after():

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

在这里创建一个宽400像素、高400像素的应用程序。它上面没有小部件。它只会显示一个框架。然后,调用self.root.after(),其中self.root是对Tk()对象的引用。after()接受两个参数:

睡眠的毫秒数

睡眠完成时要调用的方法

在这种情况下,应用程序将在3秒钟后将字符串打印到标准输出。您可以将after()视为tkinter版本的time.sleep(),但它还增加了在睡眠结束后调用函数的功能。

您可以使用此功能来改善用户体验。通过添加Pythonsleep()调用,可以使应用程序看起来加载更快,然后在启动后启动一些运行时间更长的进程。那样的话,用户无需等待应用程序打开。

睡眠在wxPython

wxPython和Tkinter之间有两个主要区别:

wxPython有更多的小部件。

wxPython的目标是在所有平台上看起来和感觉都是本地的。

Python中没有wxPython框架,所以您需要自己安装它。如果您不熟悉wxPython,请查看如何使用wxPython构建Python GUI应用程序。

在wxPython中,您可以使用wx.CallLater()添加Python调用:

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

在这里,您可以直接将wx.Frame子类化,然后调用wx.CallLater()。此函数采用与Tkinter的after()相同的参数:

睡眠的毫秒数

睡眠完成时调用的方法

运行此代码时,应该会看到一个没有任何小部件的小空白窗口出现。4秒钟后,您将看到字符串'I was delayed'打印到stdout。

使用wx.CallLater()的好处之一是线程安全。您可以在线程中使用此方法来调用wxPython主应用程序中的函数。

结论

通过本教程,您获得了一项有价值的新技术,可以添加到Python工具箱中!您知道如何添加延迟来加快应用程序的速度,并防止它们耗尽系统资源。您甚至可以使用Pythonsleep()调用来帮助更有效地重新绘制GUI代码。这将为您的客户提供更好的用户体验!

·点赞 ·收藏

分享:

Python中的变量 我们如何定义Java?


【本文地址】


今日新闻


推荐新闻


CopyRight 2018-2019 办公设备维修网 版权所有 豫ICP备15022753号-3