摘要：Python 有几种办法能够定时调度一个工作，这就是咱们将在本文中学习的内容。
本文分享自华为云社区《Python中应用定时调度工作（Schedule Jobs）的5种形式）》，作者： Regan Yue 。
明天构建的大多数应用程序都须要某种形式的调度机制。轮询 API 或数据库、一直查看零碎健康状况、将日志存档等是常见的例子。 Kubernetes和Apache Mesos等应用主动伸缩扩容技术（Auto-scaling）的软件须要查看部署的应用程序的状态，为此它们应用定期运行的存活探针（Liveness Probe）。调度工作须要与业务逻辑解耦，因而咱们要应用解耦的执行队列，例如Redis队列。
Python 有几种办法能够定时调度一个工作，这就是咱们将在本文中学习的内容。我将应用以下形式探讨调度工作：
- 简略循环 （Simple Loops）
- 简略循环然而应用了线程 （Simple Loops but Threaded）
- 调度库 （Schedule Library）
- Python Crontab
- RQ 调度器作为解耦队列 （RQ Scheduler as decoupled queues）
简略循环 Simple loops
应用简略循环来实现调度工作这是毫不费力的。应用有限运行的 while 循环定期调用函数可用于调度作业，但这不是最好的办法，不过它是很无效的。能够应用内置time模块的slleep()来提早执行。不过这并不是大多数作业的调度形式，因为，它看起来很难看，而且与其余办法相比，它的可读性较差。
import timedef task(): print("Job Completed!") while 1: task() time.sleep(10)当波及到每天早上 9:00 或每周三早晨 7:45 等这些日程安排时，事件就变得比拟辣手了。
import datetimedef task(): print("Job Completed!") while 1: now = datetime.datetime.now() # schedule at every wednesday,7:45 pm if now.weekday == 3 and now.strftime("%H:%m") == "19:45": task() # sleep for 6 days time.sleep(6 * 24 * 60 * 60)这是我的第一工夫想到的解决办法，不用谢！这种办法的一个问题是这里的逻辑是阻塞的，即一旦在 python 我的项目中发现这段代码，它就会卡在 while 1 循环中，从而阻塞其余代码的执行。
简略循环然而应用了线程Simple loops but threaded
线程是计算机科学中的一个概念。具备本人指令的小程序由过程执行并独立治理，这就能够解决咱们第一种办法的阻塞状况，让咱们看看怎么样。
import timeimport threadingdef task(): print("Job Completed!")def schedule(): while 1: task() time.sleep(10)# makes our logic non blockingthread = threading.Thread(target=schedule)thread.start()线程启动后，其底层逻辑无奈被主线程批改，因而咱们可能须要增加资源，程序通过这些资源能够查看特定场景并依据它们执行逻辑。
定时调度库 Schedule Library
早些时候，我说应用 while 循环进行调度看起来很俊俏，调度库能够解决这个问题。
import scheduleimport timedef task(): print("Job Executing!")# for every n minutesschedule.every(10).minutes.do(task)# every hourschedule.every().hour.do(task)# every daya at specific timeschedule.every().day.at("10:30").do(task)# schedule by name of dayschedule.every().monday.do(task)# name of day with timeschedule.every().wednesday.at("13:15").do(task)while True: schedule.run_pending() time.sleep(1)正如您所见，通过这样咱们能够毫不费力地创立多个调度打算。我特地喜爱创立作业的形式和办法链（Method Chaining），另一方面，这个片段有一个 while 循环，这意味着代码被阻塞，不过我置信你曾经晓得什么能够帮忙咱们解决这个问题。
Python Crontab
Liunx 中的 crontab 实用程序是一种易于应用且被宽泛承受的调度解决方案。Python 库python-crontab提供了一个 API 来应用 Python 中的 CLI 工具。在crontab中，一个定时调度应用 unix-cron字符串格局（ *）来形容，它是一组五个值的一条线，这表明当作业应该被执行时，python-crontab 将在文件中写入 crontab 的打算转换为写入编程办法。
from crontab import CronTabcron = CronTab(user='root')job = cron.new(command='my_script.sh')job.hour.every(1)cron.write()python-crontab 不会主动保留打算，须要执行 write() 办法来保留打算。还有更多功能，我强烈建议您查看他们的文档。
RQ 调度器 RQ Scheduler
有些工作不能立刻执行，因而咱们须要依据 LIFO 或 FIFO 等队列零碎创立工作队列并弹出工作。python-rq容许咱们做到这一点，应用 Redis 作为代理来排队作业。新作业的条目存储为带有信息的哈希映射，例如created_at, enqueued_at, origin, data, description.
排队工作由名为 worker 的程序执行。workers 在 Redis 缓存中也有一个条目，负责将工作出列以及更新 Redis 中的工作状态。工作能够在须要时排队，但要安顿它们，咱们须要rq-scheduler。
from rq_scheduler import Schedulerqueue = Queue('circle', connection=Redis())scheduler = Scheduler(queue=queue)scheduler.schedule( scheduled_time=datetime.utcnow(), # Time for first execution, in UTC timezone func=func, # Function to be queued args=[arg1, arg2], # Arguments passed into function when executed kwargs={'foo': 'bar'}, # Keyword arguments passed into function when executed interval=60, # Time before the function is called again, in seconds repeat=None, # Repeat this number of times (None means repeat forever) meta={'foo': 'bar'} # Arbitrary pickleable data on the job itself)RQ worker（RQ 工作器）必须在终端中独自启动或通过 python-rq 工作器启动。一旦工作被触发，就能够在工作终端中看到，在胜利和失败场景中都能够应用独自的函数回调。
总结 Conclusion
还有一些用于调度的库，但在这里，我曾经探讨了最常见的库。值得一提的是Celery，celery 的另一个长处是用户能够在多个代理之间进行抉择。我很感谢你读到最初。也能够看看我的其余文章。干杯!
翻译起源： https://python.plainenglish.i...
点击关注，第一工夫理解华为云陈腐技术~