今天我們來(lái)介紹一下Python當(dāng)中的定時(shí)任務(wù)，主要用到的模塊是sched，當(dāng)然除了該模塊之外，還有其他的例如ApScheduler，但是相比較與sched模塊而言，后者還需要另外通過(guò)pip命令來(lái)下載，稍顯麻煩。那么在這篇教程當(dāng)中我們來(lái)講講sched模塊的使用教程。

牛刀小試

我們先來(lái)看下面這個(gè)案例，代碼如下

import sched
import time

def say_hello(name):
    print(f"Hello, world, {name}")

scheduler = sched.scheduler(time.time, time.sleep)

scheduler.enter(5, 1, say_hello, ("張三", ))
scheduler.run()

那么上述的代碼中，第一步首先則是實(shí)例化一個(gè)定時(shí)器，通過(guò)如下的代碼

import sched

scheduler = sched.scheduler()

接下來(lái)我們通過(guò)enter()方法來(lái)執(zhí)行定時(shí)任務(wù)的操作，其中的參數(shù)分別是延遲的時(shí)間、任務(wù)的優(yōu)先級(jí)以及具體的執(zhí)行函數(shù)和執(zhí)行函數(shù)中的參數(shù)。像如上的代碼就會(huì)在延遲5秒鐘之后執(zhí)行say_hello()函數(shù)

當(dāng)然要是延遲的時(shí)間相等的時(shí)候，我們可以設(shè)置任務(wù)執(zhí)行的優(yōu)先級(jí)來(lái)指定函數(shù)方法運(yùn)行的順序，例如有如下的代碼

import sched
import time

def say_hello(name):
    print(f"Hello, world, {name}")

def say_hello_2(name):
    print(f"Hello, {name}")

scheduler = sched.scheduler(time.time, time.sleep)

scheduler.enter(5, 2, say_hello, ("張三", ))
scheduler.enter(5, 1, say_hello_2, ("李四", ))
scheduler.run()

如上述代碼，盡管延遲的時(shí)間都是一樣的，但是say_hello()方法的優(yōu)先級(jí)明顯要比say_hello_2()方法要低一些，因此后者會(huì)優(yōu)先執(zhí)行。

進(jìn)階使用

除了讓函數(shù)延遲執(zhí)行，我們還可以讓其重復(fù)執(zhí)行，具體這樣來(lái)操作，代碼如下

import sched
import time

def say_hello():
    print("Hello, world!")

scheduler = sched.scheduler(time.time, time.sleep)

def repeat_task():
    scheduler.enter(5, 1, say_hello, ())
    scheduler.enter(5, 1, repeat_task, ())

repeat_task()
scheduler.run()

這里我們新建了一個(gè)repeat_task()自定義函數(shù)，調(diào)用了scheduler.enter()方法5秒鐘執(zhí)行一次之前定義的say_hello()函數(shù)

在固定時(shí)間執(zhí)行任務(wù)

同時(shí)我們還可以讓任務(wù)在指定的時(shí)間執(zhí)行，這里用到scheduler.entertabs()方法，代碼如下

import sched
import time

def say_hello():
    print("Hello, world!")

scheduler = sched.scheduler(time.time, time.sleep)

# 指定時(shí)間執(zhí)行任務(wù)
specific_time = time.time() + 5  # 距離現(xiàn)在的5秒鐘之后執(zhí)行
scheduler.enterabs(specific_time, 1, say_hello, ())

scheduler.run()

我們傳入其中參數(shù)使其在指定的時(shí)間，也就是距離當(dāng)下的5秒鐘之后來(lái)執(zhí)行任務(wù)

執(zhí)行多個(gè)任務(wù)

這里仍然是調(diào)用enter()方法來(lái)運(yùn)行多個(gè)任務(wù)，代碼如下

import sched
import time

def task_one():
    print("Task One - Hello, world!")
    
def task_two():
    print("Task Two - Hello, world!")

scheduler = sched.scheduler(time.time, time.sleep)

# 任務(wù)一在兩秒鐘只有執(zhí)行
scheduler.enter(2, 1, task_one, ())

# 任務(wù)二在五秒鐘之后運(yùn)行
scheduler.enter(5, 1, task_two, ())

scheduler.run()

這里定義了兩個(gè)函數(shù)，task_one和task_two里面分是同樣的執(zhí)行邏輯，打印出“Hello, world!”，然后task_one()是在兩秒鐘之后執(zhí)行而task_two()則是在5秒鐘之后執(zhí)行，兩者執(zhí)行的優(yōu)先級(jí)都是一樣的。

以不同的優(yōu)先級(jí)執(zhí)行不同的任務(wù)

這回我們給task_one()和task_two()賦予不同的優(yōu)先級(jí)，看一看執(zhí)行的結(jié)果如下

import sched
import time

def task_one():
    print("Task One - Hello, world!")
    
def task_two():
    print("Task Two - Hello, world!")

scheduler = sched.scheduler(time.time, time.sleep)

# 優(yōu)先級(jí)是1
scheduler.enter(2, 2, task_one, ())

# 優(yōu)先級(jí)是2
scheduler.enter(5, 1, task_two, ())

scheduler.run()

output

Task One - Hello, world!
Task Two - Hello, world!

上述的代碼會(huì)在停頓兩秒之后運(yùn)行task_one()函數(shù)，再停頓3秒之后執(zhí)行task_two()函數(shù)

定時(shí)任務(wù)加上取消方法

我們給定時(shí)任務(wù)添加上取消的方法，代碼如下

import sched
import time

def task_one():
    print("Task One - Hello, world!")
    
def task_two():
    print("Task Two - Hello, world!")

scheduler = sched.scheduler(time.time, time.sleep)

# 任務(wù)一在兩秒鐘只有執(zhí)行
task_one_event = scheduler.enter(2, 1, task_one, ())

# 任務(wù)二在五秒鐘之后運(yùn)行
task_two_event = scheduler.enter(5, 1, task_two, ())

# 取消執(zhí)行task_one
scheduler.cancel(task_one_event)

scheduler.run()

我們將兩秒鐘之后執(zhí)行的task_one()方法給取消掉，最后就只執(zhí)行了task_two()方法，也就打印出來(lái)“Task Two - Hello, world!”

執(zhí)行備份程序

我們來(lái)寫(xiě)一個(gè)備份的腳本，在每天固定的時(shí)間將文件備份，代碼如下

import sched
import time
import shutil

def backup_files():
    source = '路徑/files'
    destination = '路徑二'
    shutil.copytree(source, destination)

def schedule_backup():
    # 創(chuàng)建新的定時(shí)器
    scheduler = sched.scheduler(time.time, time.sleep)

    # 備份程序在每天的1點(diǎn)來(lái)執(zhí)行
    backup_time = time.strptime('01:00:00', '%H:%M:%S')
    backup_event = scheduler.enterabs(time.mktime(backup_time), 1, backup_files, ())

    # 開(kāi)啟定時(shí)任務(wù)
    scheduler.run()

schedule_backup()

我們通過(guò)shutil模塊當(dāng)中的copytree()方法來(lái)執(zhí)行拷貝文件，然后在每天的1點(diǎn)準(zhǔn)時(shí)執(zhí)行

執(zhí)行定時(shí)分發(fā)郵件的程序

最后我們來(lái)執(zhí)行定時(shí)分發(fā)郵件的程序，代碼如下

import sched
import time
import smtplib
from email.mime.text import MIMEText

def send_email(subject, message, from_addr, to_addr, smtp_server):
    # 郵件的主體信息
    email = MIMEText(message)
    email['Subject'] = subject
    email['From'] = from_addr
    email['To'] = to_addr

    # 發(fā)郵件
    with smtplib.SMTP(smtp_server) as server:
        server.send_message(email)

def send_scheduled_email(subject, message, from_addr, to_addr, smtp_server, scheduled_time):
    # 創(chuàng)建定時(shí)任務(wù)的示例
    scheduler = sched.scheduler(time.time, time.sleep)

    # 定時(shí)郵件
    scheduler.enterabs(scheduled_time, 1, send_email, argument=(subject, message, from_addr, to_addr, smtp_server))

    # 開(kāi)啟定時(shí)器
    scheduler.run()

subject = 'Test Email'
message = 'This is a test email'
from_addr = 'test@example.com'
to_addr = 'test@example.com'
smtp_server = 'smtp.test.com'

scheduled_time = time.time() + 60 # 一分鐘之后執(zhí)行程序
send_scheduled_email(subject, message, from_addr, to_addr, smtp_server, scheduled_time)

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

您可能感興趣的文章: