????????在日常工作中，常常會用到需要周期性執(zhí)行的任務，一種方式是采用 Linux 系統(tǒng)自帶的 crond 結(jié)合命令行實現(xiàn)。另外一種方式是直接使用Python。????????

????????當每隔一段時間就要執(zhí)行一段程序，或者往復循環(huán)執(zhí)行某一個任務，這就需要使用定時任務來執(zhí)行程序。比如在實現(xiàn)對某個目標進行爬蟲的話，需要用到實時任務。

python中常用的定時任務主要有以下8中方法：

while True：+sleep()

threading.Timer定時器

Timeloop庫執(zhí)行定時任務

調(diào)度模塊sched

調(diào)度模塊schedule

任務框架APScheduler

分布式消息系統(tǒng)celery執(zhí)行定時任務

使用windows自帶的定時任務

接下來分別用上述8中方式來完成下面定義的Task()任務，示例代碼如下：

from datetime import datetimedef task():now = datetime.now()ts = now.strftime("%Y-%m-%d %H:%M:%S")print(ts)

1、利用while True：+sleep()實現(xiàn)定時任務

????????最簡單的方式應該就是使用time模塊來實現(xiàn)定時任務，在循環(huán)里面放入要執(zhí)行的任務，然后sleep一段時間再執(zhí)行。實現(xiàn)令當前執(zhí)行的線程暫停 n秒后再繼續(xù)執(zhí)行。所謂暫停，即令當前線程進入阻塞狀態(tài)，當達到 sleep() 函數(shù)規(guī)定的時間后，再由阻塞狀態(tài)轉(zhuǎn)為就緒狀態(tài)，等待 CPU 調(diào)度。

示例代碼：

from datetime import datetime import timedef task():now = datetime.now()ts = now.strftime("%Y-%m-%d %H:%M:%S")print(ts)def func():while True:task()time.sleep(3)func()

運行結(jié)果：

優(yōu)缺點：只能實現(xiàn)同步任務，無法執(zhí)行異步任務。執(zhí)行起來雖然是比較簡單，但不容易控制，而且sleep是個阻塞函數(shù)。只能設定間隔，不能指定具體的時間點。

2、利用threading.Timer()定時器實現(xiàn)定時任務

????????timer最基本理解就是定時器，可以啟動多個定時任務，這些定時器任務是異步執(zhí)行，所以不存在等待順序執(zhí)行問題。

Timer方法說明

Timer(interval, function, args=None, kwargs=None)	創(chuàng)建定時器
cancel()	取消定時器
start()	使用線程方式執(zhí)行
join(self, timeout=None)	等待線程執(zhí)行結(jié)束

示例代碼：

from datetime import datetime from threading import Timerdef task():now = datetime.now()ts = now.strftime("%Y-%m-%d %H:%M:%S")print(ts)def func():task()t = Timer(3, func)t.start()func()

運行結(jié)果：

優(yōu)缺點：可以實現(xiàn)異步任務，是非阻塞的，但當運行次數(shù)過多時，會出現(xiàn)報錯：Pyinstaller maximum recursion depth exceeded Error Resolution 達到最大遞歸深度，然后想到的是修改最大遞歸深度，

sys.setrecursionlimit(100000000)

但是運行到達到最大CPU時，python會直接銷毀程序。

關于更多timer用法，詳見博文：threading.Timer()定時器實現(xiàn)定時任務_IT之一小佬的博客-CSDN博客

3、使用Timeloop庫執(zhí)行定時任務

????????Timeloop是一個庫，可用于運行多周期任務。這是一個簡單的庫，使用decorator模式在線程中運行標記函數(shù)。

示例代碼：

from datetime import datetime, timedelta from timeloop import Timelooptl = Timeloop()def task():now = datetime.now()ts = now.strftime("%Y-%m-%d %H:%M:%S")print(ts + '333!')def task2():now = datetime.now()ts = now.strftime("%Y-%m-%d %H:%M:%S")print(ts + "555555!")@tl.job(interval=timedelta(seconds=2)) def sample_job_every_2s():task()@tl.job(interval=timedelta(seconds=5)) def sample_job_every_5s():task2()

關于更多timeloop用法，詳見博文：???python中定時任務timeloop庫用法詳解_IT之一小佬的博客-CSDN博客

4、利用調(diào)度模塊sched實現(xiàn)定時任務

????????sched是一種調(diào)度（延時處理機制）。sched模塊實現(xiàn)了一個通用事件調(diào)度器，在調(diào)度器類使用一個延遲函數(shù)等待特定的時間，執(zhí)行任務。同時支持多線程應用程序，在每個任務執(zhí)行后會立刻調(diào)用延時函數(shù)，以確保其他線程也能執(zhí)行。

scheduler對象主要方法：

• enter(delay, priority, action, argument)，安排一個事件來延遲delay個時間單位。
• cancel(event)：從隊列中刪除事件。如果事件不是當前隊列中的事件，則該方法將跑出一個ValueError。
• run()：運行所有預定的事件。這個函數(shù)將等待(使用傳遞給構(gòu)造函數(shù)的delayfunc()函數(shù))，然后執(zhí)行事件，直到不再有預定的事件。

示例代碼：

import sched import time from datetime import datetime# 初始化sched模塊的scheduler類 # 第一個參數(shù)是一個可以返回時間戳的函數(shù)，第二個參數(shù)可以在定時未到達之前阻塞。 schedule = sched.scheduler(time.time, time.sleep)def task(inc):now = datetime.now()ts = now.strftime("%Y-%m-%d %H:%M:%S")print(ts)schedule.enter(inc, 0, task, (inc,))def func(inc=3):# enter四個參數(shù)分別為：# 間隔事件、優(yōu)先級（用于同時間到達的兩個事件同時執(zhí)行時定序）、被調(diào)用觸發(fā)的函數(shù)、給該觸發(fā)函數(shù)的參數(shù)（tuple形式）schedule.enter(0, 0, task, (inc,))schedule.run()func()

運行結(jié)果：

關于更多sched用法，詳見博文：??https://blog.csdn.net/weixin_44799217/article/details/127353545

5、利用調(diào)度模塊schedule實現(xiàn)定時任務

????????schedule是一個第三方輕量級的任務調(diào)度模塊，可以按照秒，分，小時，日期或者自定義事件執(zhí)行時間。
????????如果想執(zhí)行多個任務，也可以添加多個task。

示例代碼：

import schedule from datetime import datetimedef task():now = datetime.now()ts = now.strftime("%Y-%m-%d %H:%M:%S")print(ts)def task2():now = datetime.now()ts = now.strftime("%Y-%m-%d %H:%M:%S")print(ts + '666!')def func():# 清空任務schedule.clear()# 創(chuàng)建一個按3秒間隔執(zhí)行任務schedule.every(3).seconds.do(task)# 創(chuàng)建一個按2秒間隔執(zhí)行任務schedule.every(2).seconds.do(task2)while True:schedule.run_pending()func()

運行結(jié)果：

優(yōu)缺點：需要和while Ture配合使用，而且占用的CPU也比其他幾種多的多，占用內(nèi)存也是較大。

關于更多schedule用法，詳見博文：?https://blog.csdn.net/weixin_44799217/article/details/127352957

6、利用任務框架ASPcheduler實現(xiàn)定時任務

????????APScheduler是Python的一個定時任務框架，用于執(zhí)行周期或者定時任務，該框架不僅可以添加、刪除定時任務，還可以將任務存儲到數(shù)據(jù)庫中，實現(xiàn)任務的持久化，使用起來非常方便。

示例代碼：

from datetime import datetime from apscheduler.schedulers.blocking import BlockingSchedulerdef task():now = datetime.now()ts = now.strftime("%Y-%m-%d %H:%M:%S")print(ts)def task2():now = datetime.now()ts = now.strftime("%Y-%m-%d %H:%M:%S")print(ts + '666!')def func():# 創(chuàng)建調(diào)度器BlockingScheduler()scheduler = BlockingScheduler()scheduler.add_job(task, 'interval', seconds=3, id='test_job1')# 添加任務，時間間隔為5秒scheduler.add_job(task2, 'interval', seconds=5, id='test_job2')scheduler.start()func()

運行結(jié)果：

關于更多apschedule用法，詳見博文：python中定時任務apscheduler庫用法詳解_IT之一小佬的博客-CSDN博客?

7、使用分布式消息系統(tǒng)celery執(zhí)行定時任務

????????Celery是一個簡單，靈活，可靠的分布式系統(tǒng)，用于處理大量消息，同時為操作提供維護此類系統(tǒng)所需的工具, 也可用于任務調(diào)度。Celery 的配置比較麻煩，如果你只是需要一個輕量級的調(diào)度工具，Celery 不會是一個好選擇。

????????Celery 是一個強大的分布式任務隊列，它可以讓任務的執(zhí)行完全脫離主程序，甚至可以被分配到其他主機上運行。我們通常使用它來實現(xiàn)異步任務（async task）和定時任務（crontab）。 異步任務比如是發(fā)送郵件、或者文件上傳, 圖像處理等等一些比較耗時的操作 ，定時任務是需要在特定時間執(zhí)行的任務。

注意：celery本身并不具備任務的存儲功能，在調(diào)度任務的時候肯定是要把任務存起來的，因此在使用celery的時候還需要搭配一些具備存儲、訪問功能的工具，比如：消息隊列、Redis緩存、數(shù)據(jù)庫等。官方推薦的是消息隊列RabbitMQ，有些時候使用Redis也是不錯的選擇。

8、使用windows自帶的定時任務

????????略。這兒不做細述！

總結(jié)

以上是生活随笔為你收集整理的python实现定时任务的8种方式详解的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。