本篇內(nèi)容介紹了“Python3定時(shí)任務(wù)的實(shí)現(xiàn)方式”的有關(guān)知識(shí)，在實(shí)際案例的操作過(guò)程中，不少人都會(huì)遇到這樣的困境，接下來(lái)就讓小編帶領(lǐng)大家學(xué)習(xí)一下如何處理這些情況吧！希望大家仔細(xì)閱讀，能夠?qū)W有所成！

創(chuàng)新互聯(lián)建站是專業(yè)的龍圩網(wǎng)站建設(shè)公司，龍圩接單;提供網(wǎng)站設(shè)計(jì)、網(wǎng)站建設(shè),網(wǎng)頁(yè)設(shè)計(jì),網(wǎng)站設(shè)計(jì),建網(wǎng)站,PHP網(wǎng)站建設(shè)等專業(yè)做網(wǎng)站服務(wù);采用PHP框架,可快速的進(jìn)行龍圩網(wǎng)站開發(fā)網(wǎng)頁(yè)制作和功能擴(kuò)展;專業(yè)做搜索引擎喜愛(ài)的網(wǎng)站,專業(yè)的做網(wǎng)站團(tuán)隊(duì),希望更多企業(yè)前來(lái)合作!

Python中常用的定時(shí)任務(wù)實(shí)現(xiàn)方式：

1>循環(huán)+sleep；
2>線程模塊中Timer類；
3>schedule模塊；
4>定時(shí)框架：APScheduler

在開始之前先設(shè)定一個(gè)任務(wù)（這樣不用依賴外部環(huán)境）：
1：定時(shí)或者定點(diǎn)監(jiān)測(cè)CPU與內(nèi)存使用率；
2：將時(shí)間，CPU，內(nèi)存使用情況保存到日志文件；

先來(lái)實(shí)現(xiàn)系統(tǒng)監(jiān)測(cè)功能：
準(zhǔn)備工作：安裝psutil：pip install psutil
功能實(shí)現(xiàn)

#psutil:獲取系統(tǒng)信息模塊，可以獲取CPU，內(nèi)存，磁盤等的使用情況
import psutil
import time
import datetime
#logfile：監(jiān)測(cè)信息寫入文件
def MonitorSystem(logfile = None):
    #獲取cpu使用情況
    cpuper = psutil.cpu_percent()
    #獲取內(nèi)存使用情況：系統(tǒng)內(nèi)存大小，使用內(nèi)存，有效內(nèi)存，內(nèi)存使用率
    mem = psutil.virtual_memory()
    #內(nèi)存使用率
    memper = mem.percent
    #獲取當(dāng)前時(shí)間
    now = datetime.datetime.now()
    ts = now.strftime('%Y-%m-%d %H:%M:%S')
    line = f'{ts} cpu:{cpuper}%, mem:{memper}%'
    print(line)
    if logfile:
        logfile.write(line)

代碼運(yùn)行結(jié)果：

2019-03-21 14:23:41 cpu:0.6%, mem:77.2%

接下來(lái)我們要實(shí)現(xiàn)定時(shí)監(jiān)測(cè)，比如3s監(jiān)測(cè)一下系統(tǒng)資源使用情況。

最簡(jiǎn)單使用方式：sleep

這種方式最簡(jiǎn)單，直接使用while+sleep就可以實(shí)現(xiàn)：

def loopMonitor():
    while True:
        MonitorSystem()
        #2s檢查一次
        time.sleep(3)
loopMonitor()

輸出結(jié)果：

2019-03-21 14:28:42 cpu:1.5%, mem:77.6%
2019-03-21 14:28:45 cpu:1.6%, mem:77.6%
2019-03-21 14:28:48 cpu:1.4%, mem:77.6%
2019-03-21 14:28:51 cpu:1.4%, mem:77.6%
2019-03-21 14:28:54 cpu:1.3%, mem:77.6%

這種方式存在問(wèn)題：只能處理單個(gè)定時(shí)任務(wù)。
又來(lái)了新任務(wù)：需要每秒監(jiān)測(cè)網(wǎng)絡(luò)收發(fā)字節(jié)，代碼實(shí)現(xiàn)如下：

def MonitorNetWork(logfile = None):
    #獲取網(wǎng)絡(luò)收信息
    netinfo = psutil.net_io_counters()
    #獲取當(dāng)前時(shí)間
    now = datetime.datetime.now()
    ts = now.strftime('%Y-%m-%d %H:%M:%S')
    line = f'{ts} bytessent={netinfo.bytes_sent}, bytesrecv={netinfo.bytes_recv}'
    print(line)
    if logfile:
        logfile.write(line)
MonitorNetWork()

代碼執(zhí)行結(jié)果：

2019-03-21 14:47:21 bytessent=169752183, bytesrecv=1107900973

如果我們同時(shí)在while循環(huán)中監(jiān)測(cè)兩個(gè)任務(wù)會(huì)有等待問(wèn)題，不能每秒監(jiān)測(cè)網(wǎng)絡(luò)情況。

Timer實(shí)現(xiàn)方式

timer最基本理解就是定時(shí)器，我們可以啟動(dòng)多個(gè)定時(shí)任務(wù)，這些定時(shí)器任務(wù)是異步執(zhí)行，所以不存在等待順序執(zhí)行問(wèn)題。
先來(lái)看Timer的基本使用：
導(dǎo)入：from threading import Timer
主要方法：

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

定時(shí)器只能執(zhí)行一次，如果需要重復(fù)執(zhí)行，需要重新添加任務(wù)；
我們先來(lái)看基本使用：

from threading import Timer
#記錄當(dāng)前時(shí)間
print(datetime.datetime.now())
#3S執(zhí)行一次
sTimer = Timer(3, MonitorSystem)
#1S執(zhí)行一次
nTimer = Timer(1, MonitorNetWork)
#使用線程方式執(zhí)行
sTimer.start()
nTimer.start()
#等待結(jié)束
sTimer.join()
nTimer.join()
#記錄結(jié)束時(shí)間
print(datetime.datetime.now())

輸出結(jié)果：

2019-03-21 15:13:36.739798
2019-03-21 15:13:37 bytessent=171337324, bytesrecv=1109002349
2019-03-21 15:13:39 cpu:1.4%, mem:93.2%
2019-03-21 15:13:39.745187

可以看到，花費(fèi)時(shí)間為3S，但是我們想要做的是每秒監(jiān)控網(wǎng)絡(luò)狀態(tài)；如何處理。
Timer只能執(zhí)行一次，所以執(zhí)行完成之后需要再次添加任務(wù)，我們對(duì)代碼進(jìn)行修改：

from threading import Timer
import psutil
import time
import datetime
def MonitorSystem(logfile = None):
    cpuper = psutil.cpu_percent()
    mem = psutil.virtual_memory()
    memper = mem.percent
    now = datetime.datetime.now()
    ts = now.strftime('%Y-%m-%d %H:%M:%S')
    line = f'{ts} cpu:{cpuper}%, mem:{memper}%'
    print(line)
    if logfile:
        logfile.write(line)
    #啟動(dòng)定時(shí)器任務(wù)，每三秒執(zhí)行一次
    Timer(3, MonitorSystem).start()

def MonitorNetWork(logfile = None):
    netinfo = psutil.net_io_counters()
    now = datetime.datetime.now()
    ts = now.strftime('%Y-%m-%d %H:%M:%S')
    line = f'{ts} bytessent={netinfo.bytes_sent}, bytesrecv={netinfo.bytes_recv}'
    print(line)
    if logfile:
        logfile.write(line)
    #啟動(dòng)定時(shí)器任務(wù)，每秒執(zhí)行一次
    Timer(1, MonitorNetWork).start()
MonitorSystem()
MonitorNetWork()

執(zhí)行結(jié)果：

2019-03-21 15:18:21 cpu:1.5%, mem:93.2%
2019-03-21 15:18:21 bytessent=171376522, bytesrecv=1109124678
2019-03-21 15:18:22 bytessent=171382215, bytesrecv=1109128294
2019-03-21 15:18:23 bytessent=171384278, bytesrecv=1109129702
2019-03-21 15:18:24 cpu:1.9%, mem:93.2%
2019-03-21 15:18:24 bytessent=171386341, bytesrecv=1109131110
2019-03-21 15:18:25 bytessent=171388527, bytesrecv=1109132600
2019-03-21 15:18:26 bytessent=171390590, bytesrecv=1109134008

從時(shí)間中可以看到，這兩個(gè)任務(wù)可以同時(shí)進(jìn)行不存在等待問(wèn)題。
Timer的實(shí)質(zhì)是使用線程方式去執(zhí)行任務(wù)，每次執(zhí)行完后會(huì)銷毀，所以不必?fù)?dān)心資源問(wèn)題。

調(diào)度模塊：schedule

schedule是一個(gè)第三方輕量級(jí)的任務(wù)調(diào)度模塊，可以按照秒，分，小時(shí)，日期或者自定義事件執(zhí)行時(shí)間；
安裝方式：

pip install schedule

我們來(lái)看一個(gè)例子：

import datetime
import schedule
import time
def func():
    now = datetime.datetime.now()
    ts = now.strftime('%Y-%m-%d %H:%M:%S')
    print('do func  time :',ts)
def func2():
    now = datetime.datetime.now()
    ts = now.strftime('%Y-%m-%d %H:%M:%S')
    print('do func2 time：',ts)
def tasklist():
    #清空任務(wù)
    schedule.clear()
    #創(chuàng)建一個(gè)按秒間隔執(zhí)行任務(wù)
    schedule.every(1).seconds.do(func)
    #創(chuàng)建一個(gè)按2秒間隔執(zhí)行任務(wù)
    schedule.every(2).seconds.do(func2)
    #執(zhí)行10S
    for i in range(10):
        schedule.run_pending()
        time.sleep(1)
tasklist()

執(zhí)行結(jié)果：

do func  time : 2019-03-22 08:51:38
do func2 time： 2019-03-22 08:51:39
do func  time : 2019-03-22 08:51:39
do func  time : 2019-03-22 08:51:40
do func2 time： 2019-03-22 08:51:41
do func  time : 2019-03-22 08:51:41
do func  time : 2019-03-22 08:51:42
do func2 time： 2019-03-22 08:51:43
do func  time : 2019-03-22 08:51:43
do func  time : 2019-03-22 08:51:44
do func2 time： 2019-03-22 08:51:45
do func  time : 2019-03-22 08:51:45
do func  time : 2019-03-22 08:51:46

執(zhí)行過(guò)程分析：

>1>因?yàn)槔县堅(jiān)趈upyter下執(zhí)行，所以先將schedule任務(wù)清空；
>2>按時(shí)間間在schedule中隔添加任務(wù)；
>3>老貓這里按照秒間隔添加func，按照兩秒間隔添加func2;
>4>schedule添加任務(wù)后，需要查詢?nèi)蝿?wù)并執(zhí)行任務(wù)；
>5>為了防止占用資源，每秒查詢到點(diǎn)任務(wù)，然后順序執(zhí)行；

第5個(gè)順序執(zhí)行怎么理解，我們修改func函數(shù)，里面添加time.sleep(2)
然后只執(zhí)行func工作，輸出結(jié)果：

do func  time : 2019-03-22 09:00:59
do func  time : 2019-03-22 09:01:02
do func  time : 2019-03-22 09:01:05

可以看到時(shí)間間隔為3S，為什么不是1S？
因?yàn)檫@個(gè)按照順序執(zhí)行，func休眠2S，循環(huán)任務(wù)查詢休眠1S，所以會(huì)存在這個(gè)問(wèn)題。
在我們使用這種方式執(zhí)行任務(wù)需要注意這種阻塞現(xiàn)象。
我們看下schedule模塊常用使用方法：

#schedule.every(1)創(chuàng)建Job, seconds.do(func)按秒間隔查詢并執(zhí)行
schedule.every(1).seconds.do(func)
#添加任務(wù)按分執(zhí)行
schedule.every(1).minutes.do(func)
#添加任務(wù)按天執(zhí)行
schedule.every(1).days.do(func)
#添加任務(wù)按周執(zhí)行
schedule.every().weeks.do(func)
#添加任務(wù)每周1執(zhí)行，執(zhí)行時(shí)間為下周一這一時(shí)刻時(shí)間
schedule.every().monday.do(func)
#每周1，1點(diǎn)15開始執(zhí)行
schedule.every().monday.at("12:00").do(job)

這種方式局限性：如果工作任務(wù)回非常耗時(shí)就會(huì)影響其他任務(wù)執(zhí)行。我們可以考慮使用并發(fā)機(jī)制配置這個(gè)模塊使用。

任務(wù)框架APScheduler

APScheduler是Python的一個(gè)定時(shí)任務(wù)框架，用于執(zhí)行周期或者定時(shí)任務(wù)，
可以基于日期、時(shí)間間隔，及類似于Linux上的定時(shí)任務(wù)crontab類型的定時(shí)任務(wù)；
該該框架不僅可以添加、刪除定時(shí)任務(wù)，還可以將任務(wù)存儲(chǔ)到數(shù)據(jù)庫(kù)中，實(shí)現(xiàn)任務(wù)的持久化，使用起來(lái)非常方便。
安裝方式：pip install apscheduler
apscheduler組件及簡(jiǎn)單說(shuō)明：

1>triggers（觸發(fā)器）：觸發(fā)器包含調(diào)度邏輯，每一個(gè)作業(yè)有它自己的觸發(fā)器
2>job stores（作業(yè)存儲(chǔ)）:用來(lái)存儲(chǔ)被調(diào)度的作業(yè)，默認(rèn)的作業(yè)存儲(chǔ)器是簡(jiǎn)單地把作業(yè)任務(wù)保存在內(nèi)存中,支持存儲(chǔ)到MongoDB，redis數(shù)據(jù)庫(kù)中
3> executors（執(zhí)行器）：執(zhí)行器用來(lái)執(zhí)行定時(shí)任務(wù)，只是將需要執(zhí)行的任務(wù)放在新的線程或者線程池中運(yùn)行
4>schedulers（調(diào)度器）：調(diào)度器是將其它部分聯(lián)系在一起,對(duì)使用者提供接口，進(jìn)行任務(wù)添加，設(shè)置，刪除。

來(lái)看一個(gè)簡(jiǎn)單例子：

import time
from apscheduler.schedulers.blocking import BlockingScheduler
def func():
    now = datetime.datetime.now()
    ts = now.strftime('%Y-%m-%d %H:%M:%S')
    print('do func  time :',ts)

def func2():
    #耗時(shí)2S
    now = datetime.datetime.now()
    ts = now.strftime('%Y-%m-%d %H:%M:%S')
    print('do func2 time：',ts)
    time.sleep(2)

def dojob():
    #創(chuàng)建調(diào)度器：BlockingScheduler
    scheduler = BlockingScheduler()
    #添加任務(wù),時(shí)間間隔2S
    scheduler.add_job(func, 'interval', seconds=2, id='test_job1')
    #添加任務(wù),時(shí)間間隔5S
    scheduler.add_job(func2, 'interval', seconds=3, id='test_job2')
    scheduler.start()
dojob()

輸出結(jié)果：

do func  time : 2019-03-22 10:32:20
do func2 time： 2019-03-22 10:32:21
do func  time : 2019-03-22 10:32:22
do func  time : 2019-03-22 10:32:24
do func2 time： 2019-03-22 10:32:24
do func  time : 2019-03-22 10:32:26

輸出結(jié)果中可以看到：任務(wù)就算是有延時(shí)，也不會(huì)影響其他任務(wù)執(zhí)行。
APScheduler框架提供豐富接口去實(shí)現(xiàn)定時(shí)任務(wù)，可以去參考官方文檔去查看使用方式。

“Python3定時(shí)任務(wù)的實(shí)現(xiàn)方式”的內(nèi)容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業(yè)相關(guān)的知識(shí)可以關(guān)注創(chuàng)新互聯(lián)網(wǎng)站，小編將為大家輸出更多高質(zhì)量的實(shí)用文章！

文章標(biāo)題：Python3定時(shí)任務(wù)的實(shí)現(xiàn)方式
URL地址：http://chinadenli.net/article32/gsjopc.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供品牌網(wǎng)站建設(shè)、軟件開發(fā)、ChatGPT、網(wǎng)站改版、品牌網(wǎng)站設(shè)計(jì)、網(wǎng)站設(shè)計(jì)公司

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如需處理請(qǐng)聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來(lái)源： 創(chuàng)新互聯(lián)