您的位置:

用Python编写定时任务:threading.timer教程

Python是一个广泛用于编程的语言,在日常开发中经常需要进行一些定时的任务,比如定时执行爬虫、发送邮件等等。实现这些功能最常用的方法是使用Python的定时任务模块,threading.timer。本篇文章将详细阐述如何使用threading.timer模块实现定时任务的功能。

一、threading.timer简介

threading.timer模块是Python中一个非常方便的模块,它提供了定时执行任务的功能。使用该模块,可以让程序按照指定的时间间隔执行任务,非常适合需要在特定时间执行某些操作的场合。其中,通过设置join()方法可以让程序等待线程执行完毕后再结束。下面是使用threading.timer的一个示例代码:
import threading
import time

def func():
    print("Hello, world!")
    
def run():
    while True:
        t = threading.Timer(30.0, func)
        t.start()
        t.join()
        time.sleep(60)

if __name__ == "__main__":
    run()
在上述代码中,func()函数是需要定时执行的任务,本例中每30秒执行一次。run()函数中,通过调用threading.Timer实例来创建定时器,然后执行任务,并通过join()方法等待任务执行完毕。函数最后调用time.sleep()方法来休眠,以等待下一次任务的执行。

二、使用threading.timer实现爬虫定时任务

在爬虫领域,有时需要定时更新爬取的数据,以保证数据的准确性。下面我们将以一个爬取天气预报的简单爬虫为例,演示如何使用threading.timer实现爬虫的定时任务:
import threading
import time
import requests
from bs4 import BeautifulSoup

def get_weather():
    url = "http://www.weather.com.cn/weather/101280101.shtml"
    res = requests.get(url)
    res.encoding = "utf-8"
    soup = BeautifulSoup(res.text, "html.parser")
    # 获取当天天气
    today_weather = soup.find("p", class_="wea").string.strip()
    # 获取实时温度
    real_time_temp = soup.find("span", class_="tem").string.strip()
    return today_weather, real_time_temp

def run():
    while True:
        # 获取当前时间
        now_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        # 执行任务
        today_weather, real_time_temp = get_weather()
        # 输出结果
        print("{}\n天气:{}\n温度:{}".format(now_time, today_weather, real_time_temp))
        time.sleep(1800)

if __name__ == "__main__":
    run()
在上述代码中,我们使用了requests和BeautifulSoup等模块来实现对天气预报网站的爬取和数据解析,然后每30分钟输出一次结果。其中,可以根据实际需求自行修改时间间隔和输出方式。

三、使用threading.timer实现邮件发送任务

在Python中,借助smtplib和email模块,我们可以很方便地实现邮件的发送功能。下面我们将演示如何使用threading.timer实现定时发送邮件:
import threading
import time
import smtplib
from email.mime.text import MIMEText
from email.header import Header

def send_email():
    # 发送邮件的基本信息
    smtp_server = "smtp.qq.com"  # SMTP邮件服务器
    sender = "sender@qq.com"  # 发件人邮箱
    receiver = "receiver@qq.com"  # 收件人邮箱
    username = "sender@qq.com"  # 发件人邮箱账号
    password = "password"  # 发件人邮箱密码

    # 构造邮件内容
    message = MIMEText("这是一封测试邮件", "plain", "utf-8")
    message["From"] = Header("发件人姓名", "utf-8")
    message["To"] = Header("收件人姓名", "utf-8")
    message["Subject"] = Header("Python测试邮件", "utf-8")
    
    try:
        smtp_obj = smtplib.SMTP_SSL(smtp_server, 465)
        smtp_obj.login(username, password)
        smtp_obj.sendmail(sender, receiver, message.as_string())
        print("邮件发送成功!")
    except smtplib.SMTPException:
        print("邮件发送失败!")

def run():
    while True:
        # 执行任务
        send_email()
        # 等待60秒
        time.sleep(60)

if __name__ == "__main__":
    run()
在上述代码中,我们构造了一封简单的测试邮件,然后每分钟执行一次send_email()任务发送邮件。其中,可以根据实际需求自行修改邮件内容和时间间隔。 综上所述,我们可以看到,使用Python中的threading.timer模块可以很方便地实现定时任务的功能,并且应用场景非常广泛。无论是爬虫定时更新数据,还是定时发送邮件,都可以借助该模块来实现。