Python에서 sched 모듈을 사용하여 예약된 작업을 구현하는 방법

빠른 테스트

먼저 다음 사례를 살펴보겠습니다. 코드는 다음과 같습니다

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

위 코드의 첫 번째 단계는 다음 코드를 통해 타이머를 인스턴스화하는 것입니다

import sched

scheduler = sched.scheduler()

그런 다음 enter( ) 메소드를 사용하여 예약된 작업을 수행합니다. 매개변수는 지연 시간, 작업 우선순위, 특정 실행 함수 및 실행 함수의 매개변수입니다. 위와 같은 코드는 5초의 지연 후에 say_hello() 함수를 실행합니다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()

물론 지연 시간이 동일하다면 작업 실행 우선 순위를 설정하여 작업 순서를 지정할 수 있습니다. 예를 들어 다음 코드가 있습니다

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)

위 코드와 같이 지연 시간은 동일하지만 say_hello() 메서드의 우선 순위는 분명히 더 높습니다. say_hello_2()의 메소드가 더 낮으므로 후자가 먼저 실행됩니다.

고급 사용🎜🎜함수 실행을 지연시키는 것 외에도 반복적으로 실행되도록 할 수도 있습니다. 구체적으로 코드는 다음과 같습니다🎜rrreee🎜여기서 새로운 repeat_task()를 만들었습니다. 사용자 정의 함수, scheduler.enter() 메서드가 호출되어 이전에 정의된 say_hello() 함수를 실행합니다. 🎜

정해진 시간에 작업 실행

🎜 동시에 지정된 시간에 작업이 실행되도록 할 수도 있습니다. 여기서는 scheduler.entertabs() 메서드를 사용합니다. 코드는 다음과 같습니다🎜rrreee🎜 매개변수를 전달합니다. 현재 시점으로부터 5년 후인 지정된 시간에 실행되도록 합니다. 초 후에 작업을 실행합니다🎜

여러 작업 수행

🎜여기서는 여전히 enter()를 호출합니다. 여러 작업을 실행하는 방법은 다음과 같습니다🎜rrreee🎜여기에는 두 가지 함수가 정의되어 있습니다. task_one과 task_two는 동일한 실행 논리를 가지고 있습니다. "Hello, world!" 출력된 후 task_one()이 두 함수 모두에 사용됩니다. 이는 초 후에 실행되고 task_two()는 5초 후에 실행됩니다. 둘 다 동일합니다. 🎜

우선순위가 다른 여러 작업 실행

🎜이번에는 task_one()과 task_two()에 서로 다른 우선순위를 부여합니다. 실행 결과를 참조하세요. 🎜rrreee🎜output🎜

🎜작업 1 - Hello, world!
작업 2 - Hello, world!🎜

🎜위 코드는 2초 후에 실행됩니다. task_one() 함수를 입력하고 3초간 기다린 후 task_two() 함수를 실행합니다🎜

예약된 작업에 취소 방법 추가

🎜예약된 작업에 취소를 추가합니다 메소드의 코드는 다음과 같습니다🎜rrreee🎜2초 후에 실행될 task_one() 메소드를 취소하고 최종적으로 task_two() 메소드만 실행합니다. "Task Two - Hello, world!"를 출력하세요🎜🎜백업 프로그램 실행🎜🎜매일 정해진 시간에 파일을 백업하도록 백업 스크립트를 작성해보겠습니다. 코드는 다음과 같습니다🎜rrreee🎜 를 사용합니다. shutdown 모듈의 copytree() 메소드는 복사된 파일을 실행한 후 매일 1시에 정시에 실행합니다🎜🎜이메일을 정기적으로 배포하는 프로그램을 실행합니다🎜🎜드디어 , 이메일을 정기적으로 배포하는 프로그램을 실행하겠습니다🎜 rrreee.

위 내용은 Python에서 sched 모듈을 사용하여 예약된 작업을 구현하는 방법의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!