Python定时任务Schedule模块介绍

互联网软件开发中可以说我们离不开定时任务的功能开发。12306支付时间限制30分钟、定时短信通知、定时统计报表、定时上线活动、数据备份等。我理解的定时任务特点：周期性、定时性、离线性、低调性（夜深人静默默无闻哈哈）。定时任务在软件开发中起到了无私奉献、不求回报的大无畏精神。今天给大家介绍Python定时任务Schedule。

Python Schedule优势：

简单易用的API用于作业调度。
非常轻量级，没有外部依赖性。
出色的测试覆盖率。
在Python 3.6、3.7、3.8和3.9上测试通过。
进程内的任务调用不再需要额外的进程。

Python Schedule缺点：

持久性的任务
毫秒级的任务
并发执行（可以用多线程解决）
本地化（时区、工作日或节假日）

1、Python Schedule模块（安装）

#官方文档https://schedule.readthedocs.io/en/stable/#安装pip install schedule

2、Python Schedule模块（Hello Schedule）

#模块导入import scheduleimport time#定时执行函数def job(): with open('schedule_history.log', 'a', encoding='utf-8') as objFile: objFile.write('Hello Schedule!!!n') #每秒循环执行schedule.every().second.do(job)#守护进程while True: schedule.run_pending() time.sleep(1)

3、Python Schedule模块（业务类）

#业务逻辑处理类class HandelDataService(object): def __init__(self): pass def run(self): with open('business.txt', 'a', encoding='utf-8') as fileobj: fileobj.write('Hello Businessn') import scheduleimport time#导入业务类from business import HandelDataService#调用类业务逻辑def handle_dispatcher(): objDataService = HandelDataService() objDataService.run() schedule.every().second.do(handle_dispatcher)while True: schedule.run_pending() time.sleep(1)

4、Python Schedule模块（后台运行Nohup）

nohup python3 background.py 2>&1 &

5、Python Schedule模块（后台运行官方推荐方式）

import threadingimport timeimport schedule#启动一个shedule pending线程def run_schedule_thread(interval=1): schedule_thread_event = threading.Event() class ScheduleThread(threading.Thread): @classmethod def run(cls): while not schedule_thread_event.is_set(): schedule.run_pending() time.sleep(interval) schedule_thread = ScheduleThread() schedule_thread.start() return schedule_thread_eventdef background_job(): with open('demo.txt', 'a', encoding='utf-8') as fileobj: fileobj.write('Hello Pythonn')#每秒执行一次定时任务schedule.every().second.do(background_job)#调用schedule pending线程run_schedule_event = run_schedule_thread()

6、Python Schedule模块（Crontab）

#每3秒执行一次schedule.every(3).seconds.do(job)#每3分钟执行一次schedule.every(3).minutes.do(job)#每3小时执行一次schedule.every(3).hours.do(job)#每3天执行一次schedule.every(3).days.do(job)#每3周执行一次schedule.every(3).weeks.do(job)#每分钟的23秒执行一次schedule.every().minute.at(":23").do(job)#每小时的42分钟执行一次schedule.every().hour.at(":42").do(job)#每5小时的20分30秒执行一次schedule.every(5).hours.at("20:30").do(job)#每天的10:30执行schedule.every().day.at("10:30").do(job)#每天的10:30:42秒执行一次schedule.every().day.at("10:30:42").do(job)#每星期一执行一次schedule.every().monday.do(job)#每星期三的13:15分执行一次schedule.every().wednesday.at("13:15").do(job)#每5天到10天随机天数执行一次schedule.every(5).to(10).seconds.do(my_job)#每小时执行一次至2030年1月1日18点schedule.every(1).hours.until("2030-01-01 18:33").do(job)

7、Python Schedule模块（业务传参）

import scheduleimport time#函数传参def job(name,address): with open('demo3.txt', 'a', encoding='utf-8') as fileobj: fileobj.write('Hello ' name ", " address 'n')#定时任务调用传值#每秒执行一次schedule.every().second.do(job,name="second",address="beijing city")while True: schedule.run_pending() time.sleep(1)

8、Python Schedule模块（接口）

#取消某个任务second_job = schedule.every().second.do(job,name="second",address="beijing city")schedule.cancel_job(second_job)#任务只运行一次def job_exec_once(): #业务类调用 #增加return返回表示任务只运行一次 return schedule.CancelJobschedule.every().second.do(job_exec_once)#获取所有任务all_jobs = schedule.get_jobs()#取消所有任务schedule.clear()#根据任务tag选择性过滤任务schedule.every().day.do(greet, 'Andrea').tag('daily-tasks', 'friend')schedule.every().hour.do(greet, 'John').tag('hourly-tasks', 'friend')schedule.every().hour.do(greet, 'Monica').tag('hourly-tasks', 'customer')schedule.every().day.do(greet, 'Derek').tag('daily-tasks', 'guest')friends = schedule.get_jobs('friend')#根据任务tag选择性取消任务schedule.every().hour.do(greet, 'John').tag('hourly-tasks', 'friend')schedule.every().hour.do(greet, 'Monica').tag('hourly-tasks', 'customer')schedule.every().day.do(greet, 'Derek').tag('daily-tasks', 'guest')schedule.clear('daily-tasks')

个人非常喜欢Schedule代码语法方式。一种适合人类逻辑思维阅读的代码设计。