前言

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

當(dāng)每隔一段時(shí)間就要執(zhí)行一段程序，或者往復(fù)循環(huán)執(zhí)行某一個(gè)任務(wù)，這就需要使用定時(shí)任務(wù)來執(zhí)行程序。比如在實(shí)現(xiàn)對某個(gè)目標(biāo)進(jìn)行爬蟲的話，需要用到實(shí)時(shí)任務(wù)。

python中常用的定時(shí)任務(wù)主要有以下8種方法：

• while True：+sleep()
• threading.Timer定時(shí)器
• Timeloop庫執(zhí)行定時(shí)任務(wù)
• 調(diào)度模塊sched
• 調(diào)度模塊schedule
• 任務(wù)框架APScheduler
• 分布式消息系統(tǒng)celery執(zhí)行定時(shí)任務(wù)
• 使用windows自帶的定時(shí)任務(wù)

接下來分別用上述8中方式來完成下面定義的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ù)

最簡單的方式應(yīng)該就是使用time模塊來實(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ù)，無法執(zhí)行異步任務(wù)。執(zhí)行起來雖然是比較簡單，但不容易控制，而且sleep是個(gè)阻塞函數(shù)。只能設(shè)定間隔，不能指定具體的時(shí)間點(diǎn)。

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

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

示例代碼：

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ù)過多時(shí)，會出現(xiàn)報(bào)錯：Pyinstaller maximum recursion depth exceeded Error Resolution 達(dá)到最大遞歸深度，然后想到的是修改最大遞歸深度，

sys.setrecursionlimit(100000000)

但是運(yùn)行到達(dá)到最大CPU時(shí)，python會直接銷毀程序。

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

Timeloop是一個(gè)庫，可用于運(yùn)行多周期任務(wù)。這是一個(gè)簡單的庫，使用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用法，詳見博文：   python中定時(shí)任務(wù)timeloop庫用法詳解

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

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

scheduler對象主要方法：

• enter(delay, priority, action, argument)，安排一個(gè)事件來延遲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類
# 第一個(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)先級（用于同時(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用法，詳見博文：http://www.dbjr.com.cn/article/272340.htm

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

schedule是一個(gè)第三方輕量級的任務(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用法，詳見博文： http://www.dbjr.com.cn/article/272345.htm

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

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

示例代碼：

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用法，詳見博文：python中定時(shí)任務(wù)apscheduler庫用法詳解

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

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

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

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

8、使用windows自帶的定時(shí)任務(wù)

略。這兒不做細(xì)述！

總結(jié)

到此這篇關(guān)于python實(shí)現(xiàn)定時(shí)任務(wù)的八種方式的文章就介紹到這了,更多相關(guān)python定時(shí)任務(wù)方式內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論