????????在日常工作中，常常會(huì)用到需要周期性執(zhí)行的任務(wù)，一種方式是采用 Linux 系統(tǒng)自帶的 crond 結(jié)合命令行實(shí)現(xiàn)。另外一種方式是直接使用Python。????????
????????當(dāng)每隔一段時(shí)間就要執(zhí)行一段程序，或者往復(fù)循環(huán)執(zhí)行某一個(gè)任務(wù)，這就需要使用定時(shí)任務(wù)來(lái)執(zhí)行程序。比如在實(shí)現(xiàn)對(duì)某個(gè)目標(biāo)進(jìn)行爬蟲(chóng)的話，需要用到實(shí)時(shí)任務(wù)。
python中常用的定時(shí)任務(wù)主要有以下8中方法：
- while True：+sleep()
- threading.Timer定時(shí)器
- Timeloop庫(kù)執(zhí)行定時(shí)任務(wù)
- 調(diào)度模塊sched
- 調(diào)度模塊schedule
- 任務(wù)框架APScheduler
- 分布式消息系統(tǒng)celery執(zhí)行定時(shí)任務(wù)
- 使用windows自帶的定時(shí)任務(wù)
接下來(lái)分別用上述8中方式來(lái)完成下面定義的Task()任務(wù)，示例代碼如下：
from datetime import datetime
def task():
now = datetime.now()
ts = now.strftime("%Y-%m-%d %H:%M:%S")
print(ts)
1、利用while True：+sleep()實(shí)現(xiàn)定時(shí)任務(wù)
????????最簡(jiǎn)單的方式應(yīng)該就是使用time模塊來(lái)實(shí)現(xiàn)定時(shí)任務(wù)，在循環(huán)里面放入要執(zhí)行的任務(wù)，然后sleep一段時(shí)間再執(zhí)行。實(shí)現(xiàn)令當(dāng)前執(zhí)行的線程暫停 n秒后再繼續(xù)執(zhí)行。所謂暫停，即令當(dāng)前線程進(jìn)入阻塞狀態(tài)，當(dāng)達(dá)到 sleep() 函數(shù)規(guī)定的時(shí)間后，再由阻塞狀態(tài)轉(zhuǎn)為就緒狀態(tài)，等待 CPU 調(diào)度。
示例代碼：
from datetime import datetime
import time
def task():
now = datetime.now()
ts = now.strftime("%Y-%m-%d %H:%M:%S")
print(ts)
def func():
while True:
task()
time.sleep(3)
func()
運(yùn)行結(jié)果：
優(yōu)缺點(diǎn)：只能實(shí)現(xiàn)同步任務(wù)，無(wú)法執(zhí)行異步任務(wù)。執(zhí)行起來(lái)雖然是比較簡(jiǎn)單，但不容易控制，而且sleep是個(gè)阻塞函數(shù)。只能設(shè)定間隔，不能指定具體的時(shí)間點(diǎn)。
2、利用threading.Timer()定時(shí)器實(shí)現(xiàn)定時(shí)任務(wù)
????????timer最基本理解就是定時(shí)器，可以啟動(dòng)多個(gè)定時(shí)任務(wù)，這些定時(shí)器任務(wù)是異步執(zhí)行，所以不存在等待順序執(zhí)行問(wèn)題。
| Timer方法 | 說(shuō)明 |
|---|---|
| Timer(interval, function, args=None, kwargs=None) | 創(chuàng)建定時(shí)器 |
| cancel() | 取消定時(shí)器 |
| start() | 使用線程方式執(zhí)行 |
| join(self, timeout=None) | 等待線程執(zhí)行結(jié)束 |
示例代碼：
from datetime import datetime
from threading import Timer
def 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()
運(yùn)行結(jié)果：
優(yōu)缺點(diǎn)：可以實(shí)現(xiàn)異步任務(wù)，是非阻塞的，但當(dāng)運(yùn)行次數(shù)過(guò)多時(shí)，會(huì)出現(xiàn)報(bào)錯(cuò)：Pyinstaller maximum recursion depth exceeded Error Resolution 達(dá)到最大遞歸深度，然后想到的是修改最大遞歸深度，
sys.setrecursionlimit(100000000)但是運(yùn)行到達(dá)到最大CPU時(shí)，python會(huì)直接銷(xiāo)毀程序。
關(guān)于更多timer用法，詳見(jiàn)博文：threading.Timer()定時(shí)器實(shí)現(xiàn)定時(shí)任務(wù)_IT之一小佬的博客-CSDN博客
3、使用Timeloop庫(kù)執(zhí)行定時(shí)任務(wù)
????????Timeloop是一個(gè)庫(kù)，可用于運(yùn)行多周期任務(wù)。這是一個(gè)簡(jiǎn)單的庫(kù)，使用decorator模式在線程中運(yùn)行標(biāo)記函數(shù)。
示例代碼：
from datetime import datetime, timedelta
from timeloop import Timeloop
tl = 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()
關(guān)于更多timeloop用法，詳見(jiàn)博文：???python中定時(shí)任務(wù)timeloop庫(kù)用法詳解_IT之一小佬的博客-CSDN博客
4、利用調(diào)度模塊sched實(shí)現(xiàn)定時(shí)任務(wù)
????????sched是一種調(diào)度（延時(shí)處理機(jī)制）。sched模塊實(shí)現(xiàn)了一個(gè)通用事件調(diào)度器，在調(diào)度器類(lèi)使用一個(gè)延遲函數(shù)等待特定的時(shí)間，執(zhí)行任務(wù)。同時(shí)支持多線程應(yīng)用程序，在每個(gè)任務(wù)執(zhí)行后會(huì)立刻調(diào)用延時(shí)函數(shù)，以確保其他線程也能執(zhí)行。
scheduler對(duì)象主要方法：
- enter(delay, priority, action, argument)，安排一個(gè)事件來(lái)延遲delay個(gè)時(shí)間單位。
- cancel(event)：從隊(duì)列中刪除事件。如果事件不是當(dāng)前隊(duì)列中的事件，則該方法將跑出一個(gè)ValueError。
- run()：運(yùn)行所有預(yù)定的事件。這個(gè)函數(shù)將等待(使用傳遞給構(gòu)造函數(shù)的delayfunc()函數(shù))，然后執(zhí)行事件，直到不再有預(yù)定的事件。
示例代碼：
import sched
import time
from datetime import datetime
# 初始化sched模塊的scheduler類(lèi)
# 第一個(gè)參數(shù)是一個(gè)可以返回時(shí)間戳的函數(shù)，第二個(gè)參數(shù)可以在定時(shí)未到達(dá)之前阻塞。
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四個(gè)參數(shù)分別為：
# 間隔事件、優(yōu)先級(jí)（用于同時(shí)間到達(dá)的兩個(gè)事件同時(shí)執(zhí)行時(shí)定序）、被調(diào)用觸發(fā)的函數(shù)、給該觸發(fā)函數(shù)的參數(shù)（tuple形式）
schedule.enter(0, 0, task, (inc,))
schedule.run()
func()
運(yùn)行結(jié)果：
關(guān)于更多sched用法，詳見(jiàn)博文：??https://blog.csdn.net/weixin_44799217/article/details/127353545
5、利用調(diào)度模塊schedule實(shí)現(xiàn)定時(shí)任務(wù)
????????schedule是一個(gè)第三方輕量級(jí)的任務(wù)調(diào)度模塊，可以按照秒，分，小時(shí)，日期或者自定義事件執(zhí)行時(shí)間。
????????如果想執(zhí)行多個(gè)任務(wù)，也可以添加多個(gè)task。
示例代碼：
import schedule
from datetime import datetime
def 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():
# 清空任務(wù)
schedule.clear()
# 創(chuàng)建一個(gè)按3秒間隔執(zhí)行任務(wù)
schedule.every(3).seconds.do(task)
# 創(chuàng)建一個(gè)按2秒間隔執(zhí)行任務(wù)
schedule.every(2).seconds.do(task2)
while True:
schedule.run_pending()
func()
運(yùn)行結(jié)果：
優(yōu)缺點(diǎn)：需要和while Ture配合使用，而且占用的CPU也比其他幾種多的多，占用內(nèi)存也是較大。
關(guān)于更多schedule用法，詳見(jiàn)博文：?https://blog.csdn.net/weixin_44799217/article/details/127352957
6、利用任務(wù)框架ASPcheduler實(shí)現(xiàn)定時(shí)任務(wù)
????????APScheduler是Python的一個(gè)定時(shí)任務(wù)框架，用于執(zhí)行周期或者定時(shí)任務(wù)，該框架不僅可以添加、刪除定時(shí)任務(wù)，還可以將任務(wù)存儲(chǔ)到數(shù)據(jù)庫(kù)中，實(shí)現(xiàn)任務(wù)的持久化，使用起來(lái)非常方便。
示例代碼：
from datetime import datetime
from apscheduler.schedulers.blocking import BlockingScheduler
def 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')
# 添加任務(wù)，時(shí)間間隔為5秒
scheduler.add_job(task2, 'interval', seconds=5, id='test_job2')
scheduler.start()
func()
運(yùn)行結(jié)果：
關(guān)于更多apschedule用法，詳見(jiàn)博文：python中定時(shí)任務(wù)apscheduler庫(kù)用法詳解_IT之一小佬的博客-CSDN博客?
7、使用分布式消息系統(tǒng)celery執(zhí)行定時(shí)任務(wù)
????????Celery是一個(gè)簡(jiǎn)單，靈活，可靠的分布式系統(tǒng)，用于處理大量消息，同時(shí)為操作提供維護(hù)此類(lèi)系統(tǒng)所需的工具, 也可用于任務(wù)調(diào)度。Celery 的配置比較麻煩，如果你只是需要一個(gè)輕量級(jí)的調(diào)度工具，Celery 不會(huì)是一個(gè)好選擇。
????????Celery 是一個(gè)強(qiáng)大的分布式任務(wù)隊(duì)列，它可以讓任務(wù)的執(zhí)行完全脫離主程序，甚至可以被分配到其他主機(jī)上運(yùn)行。我們通常使用它來(lái)實(shí)現(xiàn)異步任務(wù)（async task）和定時(shí)任務(wù)（crontab）。 異步任務(wù)比如是發(fā)送郵件、或者文件上傳, 圖像處理等等一些比較耗時(shí)的操作 ，定時(shí)任務(wù)是需要在特定時(shí)間執(zhí)行的任務(wù)。
注意：celery本身并不具備任務(wù)的存儲(chǔ)功能，在調(diào)度任務(wù)的時(shí)候肯定是要把任務(wù)存起來(lái)的，因此在使用celery的時(shí)候還需要搭配一些具備存儲(chǔ)、訪問(wèn)功能的工具，比如：消息隊(duì)列、Redis緩存、數(shù)據(jù)庫(kù)等。官方推薦的是消息隊(duì)列RabbitMQ，有些時(shí)候使用Redis也是不錯(cuò)的選擇。文章來(lái)源：http://www.zghlxwxcb.cn/news/detail-408320.html
8、使用windows自帶的定時(shí)任務(wù)
????????略。這兒不做細(xì)述！文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-408320.html
到了這里，關(guān)于python實(shí)現(xiàn)定時(shí)任務(wù)的8種方式詳解的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
