总结：Python有几种方式来安排任务，这就是我们将在本文中学习的内容。本文分享自华为云社区《Python中使用定时调度任务（Schedule Jobs）的5种方式）》，作者：ReganYue。今天构建的大多数应用程序都需要某种形式的调度机制。轮询API或数据库、不断检查系统健康状况、归档日志等都是常见的例子。Kubernetes和ApacheMesos等自动伸缩软件需要检查已部署应用程序的状态，为此他们使用定期运行的活性探测器（LivenessProbe）。调度任务需要和业务逻辑解耦，所以我们需要使用解耦的执行队列，比如Redis队列。Python有几种定期安排任务的方法，这就是我们将在本文中学习的内容。我将使用以下方法讨论调度任务：简单循环简单循环但线程化调度库PythonCrontabRQ调度程序作为解耦队列简单循环简单循环使用简单循环来实现毫不费力的调度任务。使用无限运行的while循环定期调用函数可用于安排作业，但这不是最好的方法，但它很有效。可以使用时间模块的内置slleep()延迟执行。这不是大多数作业的安排方式，因为，嗯，它看起来很难看，而且它的可读性不如其他方法。importtimedeftask():print("JobCompleted!")while1:task()time.sleep(10)当涉及到每天早上9:00或每周三晚上7:45等时间表时，事情变得更加棘手。importdatetimedeftask():print("JobCompleted!")while1:now=datetime.datetime.now()#scheduleateverywednesday,7:45pmifnow.weekday==3和now.strftime("%H:%m")=="19:45":task()#sleepfor6daystime.sleep(6*24*60*60)这是我的第一个解决方案，不需要，谢谢！这种方法的一个问题是这里的逻辑是阻塞的，即一旦在python项目中找到这段代码，它就会卡在while1循环中，阻塞其他代码的执行。SimpleloopsbutthreadedSimpleloopsbutthreadedthread是计算机科学中的一个概念。自带指令的小程序由进程执行，独立管理，这样就可以解决我们第一种方法的阻塞情况，下面看看如何实现。importtimeimportthreadingdeftask():print("JobCompleted!")defschedule():while1:task()time.sleep(10)#让我们的逻辑非阻塞thread=threading.Thread(target=schedule)thread.start()线程启动后，它的底层逻辑不能被主线程修改，所以我们可能需要添加资源，程序可以通过这些资源检查特定的场景并根据它们执行逻辑。ScheduleLibrary前面说了用while循环调度很难看，schedulelibrary可以解决这个问题。importscheduleimporttimedeftask():print("JobExecuting!")#foreverynminutesschedule.every(10).minutes.do(task)#everyhoursschedule.every().hour.do(task)#每天在特定时间schedule.every().day.at("10:30").do(task)#按日期名称安排schedule.every().monday.do(task)#日期名称withtimeschedule.every().wednesday.at("13:15").do(task)whileTrue:schedule.run_pending()time.sleep(1)如你所见，通过这样做我们可以毫不费力地创建多个时间表。我特别喜欢创建作业的方式和方法链接，另一方面，这个片段有一个while循环，这意味着代码被阻塞了，但我相信你已经知道什么可以帮助我们解决这个问题。来自PythonCrontabLiunx的crontab实用程序是一种易于使用且被广泛接受的调度解决方案。Python库python-crontab提供了一个API以在Python中使用CLI工具。在crontab中，使用unix-cron字符串格式（*）来描述一个定时调度，它是一行的一组五个值，表示应该在什么时候执行作业，python-crontab会在文件中写入crontab将程序化转换为编写程序化方法。fromcrontabimportCronTabcron=CronTab(user='root')job=cron.new(command='my_script.sh')job.hour.every(1)cron.write()python-crontab不会自动要保存计划，您需要实现write()方法来保存计划。还有更多功能，我强烈建议您查看他们的文档。RQSchedulerRQScheduler有些任务不能立即执行，所以我们需要创建一个任务队列，并基于后进先出或先进先出等队列系统弹出任务。python-rq允许我们这样做，使用Redis作为代理来排队作业。新作业的条目存储为哈希映射，其中包含诸如created_at、enqueued_at、origin、data、description等信息。排队的任务由名为workers的程序执行。Workers在Redis缓存中也有一个条目，负责任务出队和更新Redis中的任务状态。任务可以在需要时排队，但要安排它们，我们需要rq-scheduler。fromrq_schedulerimportSchedulerqueue=Queue('circle',connection=Redis())scheduler=Scheduler(queue=queue)scheduler.schedule(scheduled_time=datetime.utcnow(),#首次执行时间，UTC时区func=func,#Functiontobequeuedargs=[arg1,arg2],#执行时传入函数的参数kwargs={'foo':'bar'},#执行时传入函数的关键字参数interval=60,#Timebefore该函数被再次调用，以秒为单位repeat=None,#重复这个次数（None意味着永远重复）meta={'foo':'bar'}#作业本身的任意pickleable数据)RQworker(RQworker)必须在终端或通过python-rqworker单独启动。一旦任务被触发，它就可以在工作终端中看到，成功和失败的场景都有单独的函数回调。结论结论还有一些用于调度的库，但在这里，我讨论了最常见的一些。值得一提的是Celery，celery的另一个优势是用户可以在多个代理之间进行选择。感谢您阅读到最后。另见我的其他文章。干杯!翻译来源：https://python.plainenglish.i...点击关注，第一时间了解华为云新技术~
