Comment utiliser le module sched en Python pour implémenter des tâches planifiées

Test rapide

Jetons d'abord un coup d'œil au cas suivant. Le code est le suivant

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()

La première étape du code ci-dessus consiste à instancier une minuterie via le code suivant

import sched

scheduler = sched.scheduler()

Ensuite, nous passons enter ( ) pour effectuer l'opération des tâches planifiées. Les paramètres sont le temps de retard, la priorité de la tâche, la fonction d'exécution spécifique et les paramètres de la fonction d'exécution. Un code comme celui ci-dessus exécutera la fonction say_hello() après un délai de 5 secondes.enter()方法来执行定时任务的操作，其中的参数分别是延迟的时间、任务的优先级以及具体的执行函数和执行函数中的参数。像如上的代码就会在延迟5秒钟之后执行say_hello()函数

当然要是延迟的时间相等的时候，我们可以设置任务执行的优先级来指定函数方法运行的顺序，例如有如下的代码

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()

如上述代码，尽管延迟的时间都是一样的，但是say_hello()方法的优先级明显要比say_hello_2()方法要低一些，因此后者会优先执行。

进阶使用

除了让函数延迟执行，我们还可以让其重复执行，具体这样来操作，代码如下

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()

这里我们新建了一个repeat_task()自定义函数，调用了scheduler.enter()方法5秒钟执行一次之前定义的say_hello()函数

在固定时间执行任务

同时我们还可以让任务在指定的时间执行，这里用到scheduler.entertabs()方法，代码如下

import sched
import time

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

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

# 指定时间执行任务
specific_time = time.time() + 5  # 距离现在的5秒钟之后执行
scheduler.enterabs(specific_time, 1, say_hello, ())

scheduler.run()

我们传入其中参数使其在指定的时间，也就是距离当下的5秒钟之后来执行任务

执行多个任务

这里仍然是调用enter()方法来运行多个任务，代码如下

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)

# 任务一在两秒钟只有执行
scheduler.enter(2, 1, task_one, ())

# 任务二在五秒钟之后运行
scheduler.enter(5, 1, task_two, ())

scheduler.run()

这里定义了两个函数，task_one和task_two里面分是同样的执行逻辑，打印出“Hello, world!”，然后task_one()是在两秒钟之后执行而task_two()则是在5秒钟之后执行，两者执行的优先级都是一样的。

以不同的优先级执行不同的任务

这回我们给task_one()和task_two()赋予不同的优先级，看一看执行的结果如下

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)

# 优先级是1
scheduler.enter(2, 2, task_one, ())

# 优先级是2
scheduler.enter(5, 1, task_two, ())

scheduler.run()

output

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

上述的代码会在停顿两秒之后运行task_one()函数，再停顿3秒之后执行task_two()函数

定时任务加上取消方法

我们给定时任务添加上取消的方法，代码如下

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)

# 任务一在两秒钟只有执行
task_one_event = scheduler.enter(2, 1, task_one, ())

# 任务二在五秒钟之后运行
task_two_event = scheduler.enter(5, 1, task_two, ())

# 取消执行task_one
scheduler.cancel(task_one_event)

scheduler.run()

我们将两秒钟之后执行的task_one()方法给取消掉，最后就只执行了task_two()方法，也就打印出来“Task Two - Hello, world!”

执行备份程序

我们来写一个备份的脚本，在每天固定的时间将文件备份，代码如下

import sched
import time
import shutil

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

def schedule_backup():
    # 创建新的定时器
    scheduler = sched.scheduler(time.time, time.sleep)

    # 备份程序在每天的1点来执行
    backup_time = time.strptime('01:00:00', '%H:%M:%S')
    backup_event = scheduler.enterabs(time.mktime(backup_time), 1, backup_files, ())

    # 开启定时任务
    scheduler.run()

schedule_backup()

我们通过shutil模块当中的copytree()

Bien sûr, si les délais sont égaux, nous pouvons définir la priorité d'exécution de la tâche pour spécifier l'ordre dans lequel les méthodes de fonction sont exécutées. , par exemple, il y a le code suivant

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

    # 发邮件
    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):
    # 创建定时任务的示例
    scheduler = sched.scheduler(time.time, time.sleep)

    # 定时邮件
    scheduler.enterabs(scheduled_time, 1, send_email, argument=(subject, message, from_addr, to_addr, smtp_server))

    # 开启定时器
    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 # 一分钟之后执行程序
send_scheduled_email(subject, message, from_addr, to_addr, smtp_server, scheduled_time)

Comme dans le code ci-dessus, bien que le temps de retard soit le même, la priorité de la méthode say_hello() est évidemment supérieure à celui de say_hello_2() La méthode est inférieure, donc cette dernière sera exécutée en premier.

Utilisation avancée🎜🎜En plus de retarder l'exécution de la fonction, nous pouvons également la faire exécuter à plusieurs reprises Plus précisément, le code est le suivant🎜rrreee🎜Ici, nous avons créé un nouveau repeat_task(). fonction personnalisée, La méthode scheduler.enter() est appelée pour exécuter la fonction say_hello() précédemment définie 🎜

Exécuter des tâches à une heure fixe

🎜 En même temps, nous pouvons également laisser la tâche être exécutée à une heure spécifiée. La méthode scheduler.entertabs() est utilisée ici. Le code est le suivant🎜rrreee🎜Nous transmettons les paramètres à. faites-le s'exécuter à l'heure spécifiée, soit dans 5 ans à partir du moment présent. Exécutez la tâche après quelques secondes🎜

Effectuer plusieurs tâches

🎜Ici, nous appelons toujours enter(). méthode pour exécuter plusieurs tâches. Le code est le suivant🎜rrreee🎜Il y a deux fonctions définies ici, task_one et task_two ont la même logique d'exécution "Bonjour tout le monde !" est imprimé, puis task_one() est utilisé dans les deux fonctions. Il sera exécuté après quelques secondes et task_two() sera exécuté après 5 secondes. les deux sont identiques. 🎜

Exécuter différentes tâches avec des priorités différentes

🎜Cette fois, nous donnons des priorités différentes à task_one() et task_two(), voir Le résultat de l'exécution est le suivant follow🎜rrreee🎜output🎜

🎜Tâche 1 - Bonjour tout le monde !
Tâche 2 - Bonjour tout le monde !🎜

🎜Le code ci-dessus s'exécutera après une pause de deux secondestask_one( ), puis attendez 3 secondes puis exécutez la fonction task_two()🎜

Ajouter la méthode d'annulation aux tâches planifiées

🎜Nous ajoutons la méthode d'annulation aux tâches planifiées, le code est le suivant🎜rrreee🎜Nous annulons la méthode task_one() qui sera exécutée après deux secondes, et finalement exécutons uniquement la méthode task_two(), c'est-à-dire Print out "Tâche deux - Bonjour tout le monde!"🎜🎜Exécutez le programme de sauvegarde🎜🎜Écrivons un script de sauvegarde pour sauvegarder le fichier à une heure fixe chaque jour🎜rrreee🎜Nous utilisons shutil<.> La méthode <code>copytree() du module exécute le fichier de copie, puis l'exécute à l'heure à 1 heure tous les jours🎜🎜Exécutez le programme de distribution régulière d'e-mails🎜🎜Enfin, nous exécutera le programme pour distribuer régulièrement des e-mails. Le code est le suivant🎜 rrreee.

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!