定时任务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(["韩明", "2981118661@qq.com"])#邮件来自地址
	msg['Subject'] = "任务调度测试邮件" #主题

	# ### 2.发送邮件 ###
	server = smtplib.SMTP_SSL("smtp.qq.com")#发送的邮件厂商 例如 smtp.168.com这是网易的
	server.login("2981118661@qq.com", "fpzrpvitpssssssf")#第二个参数就是 服务授权码在邮件设置中获取  这里授权码是错误的自己获取自己的
	server.sendmail("2981118661@qq.com", "738486932@qq.com", 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)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59