定时任务Apscheduler实践案例

定时任务Apscheduler实践案例

参考文章

https://blog.csdn.net/weixin_44799217/article/details/127353134

实现案例

本案例是使用定时任务apscheduler实现的每个三分钟发送一次邮件的任务

实现代码

import time
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.schedulers.background import BackgroundScheduler
import smtplib
from email.mime.text import MIMEText
from email.utils import formataddr# 常见的两种调度器 以及 两种触发器 以及 邮件发送# def job():
# 	print('job 3s')
# 	time.sleep(5)# def job2():
# 	print('job2 3s')
# 	time.sleep(5)# 邮件发送方法
def sendingAnEmail():# ### 1.邮件内容配置 ###msg = MIMEText("你好兄弟,这是一个间隔三分钟一次的邮件任务", 'html', 'utf-8')#邮件内容msg['From'] = formataddr(["韩明", "[e-mail]"])#邮件来自地址msg['Subject'] = "任务调度测试邮件" #主题# ### 2.发送邮件 ###server = smtplib.SMTP_SSL("smtp.qq.com")#发送的邮件厂商 例如 smtp.168.com这是网易的server.login("[e-mail]", "fpzrpvitpssssssf")#第二个参数就是 服务授权码在邮件设置中获取  这里授权码是错误的自己获取自己的server.sendmail("[e-mail]", "[e-mail]", msg.as_string())server.quit()# 当前脚本程序主入口  从这里开始执行
if __name__ == '__main__':#BlockingScheduler: 调用start函数后会阻塞当前线程。当调度器是你应用中唯一要运行的东西时（如上例）使用。sched = BlockingScheduler(timezone='MST')#例子所示，3s时间到达后，并不会“重新启动一个job线程”，而是会跳过该次调度，等到下一个周期（再等待3s），又重新调度job()。
#为了能让多个job()同时运行，可以配置调度器的参数max_instances，如下例，允许2个job()同时运行job_defaults = {'max_instances': 2}#BackgroundScheduler: 调用start后主线程不会阻塞。当你不运行任何其他框架时使用，并希望调度器在你应用的后台执行。# sched = BackgroundScheduler(timezone='MST',job_defaults=job_defaults)# sched.add_job(job, 'interval', id='3_second_job', seconds=3)# sched.add_job(job2, 'interval', id='4_second_job', seconds=3)# interval触发器sched.add_job(sendingAnEmail, 'interval', id='4_second_job', minutes=3)# cron触发器# scheduler.add_job(task, 'cron', month='1-3,7-9', day_of_week='1-2', hour='0-3', id='test_job1')sched.start()while True:print('main 1s')time.sleep(1)