简单强大的Python库!Schedule实用的周期任务调度工具
如果你想在Linux服务器上周期性地执行某个 Python 脚本,最出名的选择应该是 Crontab 脚本,但是 Crontab 具有以下缺点:
1.不方便执行秒级的任务。
2.当需要执行的定时任务有上百个的时候,Crontab的管理就会特别不方便。
另外一个选择是 Celery,但是 Celery 的配置比较麻烦,如果你只是需要一个轻量级的调度工具,Celery 不会是一个好选择。
在你想要使用一个轻量级的任务调度工具,而且希望它尽量简单、容易使用、不需要外部依赖,最好能够容纳 Crontab 的所有基本功能,那么 Schedule 模块是你的不二之选。
使用它来调度任务可能只需要几行代码,感受一下:# Python 实用宝典
import schedule
import time
def job:
print("I"m working...")
schedule.every(10).minutes.do(job)
while True:
schedule.run_pending
time.sleep(1)
上面的代码表示每10分钟执行一次 job 函数,非常简单方便。你只需要引入 schedule 模块,通过调用 scedule.every(时间数).时间类型.do(job)发布周期任务。
发布后的周期任务需要用 run_pending函数来检测是否执行,因此需要一个While循环不断地轮询这个函数。
下面具体讲讲Schedule模块的安装和初级、进阶使用方法。
1.准备
请选择以下任一种方式输入命令安装依赖:
1. Windows 环境 打开 Cmd (开始-运行-CMD)。
2. MacOS 环境 打开 Terminal (command+空格输入Terminal)。
3. 如果你用的是 VSCode编辑器 或 Pycharm,可以直接使用界面下方的Terminal.pip install schedule
2.基本使用
最基本的使用在文首已经提到过,下面给大家展示更多的调度任务例子:# Python 实用宝典
import schedule
import time
def job:
print("I"m working...")
# 每十分钟执行任务
schedule.every(10).minutes.do(job)
# 每个小时执行任务
schedule.every.hour.do(job)
# 每天的10:30执行任务
schedule.every.day.at("10:30").do(job)
# 每个月执行任务
schedule.every.monday.do(job)
# 每个星期三的13:15分执行任务
schedule.every.wednesday.at("13:15").do(job)
# 每分钟的第17秒执行任务
schedule.every.minute.at(":17").do(job)
while True:
schedule.run_pending
time.sleep(1)
可以看到,从月到秒的配置,上面的例子都覆盖到了。不过如果你想只运行一次任务的话,可以这么配:# Python 实用宝典
import schedule
import time
def job_that_executes_once:
# 此处编写的任务只会执行一次...
return schedule.CancelJob
schedule.every.day.at("22:30").do(job_that_executes_once)
while True:
schedule.run_pending
time.sleep(1)
参数传递
如果你有参数需要传递给作业去执行,你只需要这么做:# Python 实用宝典
import schedule
def greet(name):
print("Hello", name)
# do 将额外的参数传递给job函数
schedule.every(2).seconds.do(greet, name="Alice")
schedule.every(4).seconds.do(greet, name="Bob")
获取目前所有的作业
如果你想获取目前所有的作业:# Python 实用宝典
import schedule
def hello:
print("Hello world")
schedule.every.second.do(hello)
all_jobs = schedule.get_jobs
取消所有作业
如果某些机制触发了,你需要立即清除当前程序的所有作业:# Python 实用宝典
import schedule
def greet(name):
print("Hello {}".format(name))
schedule.every.second.do(greet)
schedule.clear
标签功能
在设置作业的时候,为了后续方便管理作业,你可以给作业打个标签,这样你可以通过标签过滤获取作业或取消作业。# Python 实用宝典
import schedule
def greet(name):
print("Hello {}".format(name))
# .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")
# get_jobs(标签):可以获取所有该标签的任务
friends = schedule.get_jobs("friend")
# 取消所有 daily-tasks 标签的任务
schedule.clear("daily-tasks")
设定作业截止时间
如果你需要让某个作业到某个时间截止,你可以通过这个方法:# Python 实用宝典
import schedule
from datetime import datetime, timedelta, time
def job:
print("Boo")
# 每个小时运行作业,18:30后停止
schedule.every(1).hours.until("18:30").do(job)
# 每个小时运行作业,2030-01-01 18:33 today
schedule.every(1).hours.until("2030-01-01 18:33").do(job)
# 每个小时运行作业,8个小时后停止
schedule.every(1).hours.until(timedelta(hours=8)).do(job)
# 每个小时运行作业,11:32:42后停止
schedule.every(1).hours.until(time(11, 33, 42)).do(job)
# 每个小时运行作业,2020-5-17 11:36:20后停止
schedule.every(1).hours.until(datetime(2020, 5, 17, 11, 36, 20)).do(job)
截止日期之后,该作业将无法运行。
立即运行所有作业,而不管其安排如何
如果某个机制触发了,你需要立即运行所有作业,可以调用 schedule.run_all:# Python 实用宝典
import schedule
def job_1:
print("Foo")
def job_2:
print("Bar")
schedule.every.monday.at("12:40").do(job_1)
schedule.every.tuesday.at("16:40").do(job_2)
schedule.run_all
# 立即运行所有作业,每次作业间隔10秒
schedule.run_all(delay_seconds=10)
3.高级使用
装饰器安排作业
如果你觉得设定作业这种形式太啰嗦了,也可以使用装饰器模式:# Python 实用宝典
from schedule import every, repeat, run_pending
import time
# 此装饰器效果等同于 schedule.every(10).minutes.do(job)
@repeat(every(10).minutes)
def job:
print("I am a scheduled job")
while True:
run_pending
time.sleep(1)
并行执行
默认情况下,Schedule 按顺序执行所有作业。其背后的原因是,很难找到让每个人都高兴的并行执行模型。
不过你可以通过多线程的形式来运行每个作业以解决此限制:# Python 实用宝典
import threading
import time
import schedule
def job1:
print("I"m running on thread %s" % threading.current_thread)
def job2:
print("I"m running on thread %s" % threading.current_thread)
def job3:
print("I"m running on thread %s" % threading.current_thread)
def run_threaded(job_func):
job_thread = threading.Thread(target=job_func)
job_thread.start
schedule.every(10).seconds.do(run_threaded, job1)
schedule.every(10).seconds.do(run_threaded, job2)
schedule.every(10).seconds.do(run_threaded, job3)
while True:
schedule.run_pending
time.sleep(1)
日志记录
Schedule 模块同时也支持 logging 日志记录,这么使用:# Python 实用宝典
import schedule
import logging
logging.basicConfig
schedule_logger = logging.getLogger("schedule")
# 日志级别为DEBUG
schedule_logger.setLevel(level=logging.DEBUG)
def job:
print("Hello, Logs")
schedule.every.second.do(job)
schedule.run_all
schedule.clear
效果如下:DEBUG:schedule:Running *all* 1 jobs with 0s delay in between
DEBUG:schedule:Running job Job(interval=1, unit=seconds, do=job, args=, kwargs={})
Hello, Logs
DEBUG:schedule:Deleting *all* jobs
异常处理
Schedule 不会自动捕捉异常,它遇到异常会直接抛出,这会导致一个严重的问题:后续所有的作业都会被中断执行,因此我们需要捕捉到这些异常。
你可以手动捕捉,但是某些你预料不到的情况需要程序进行自动捕获,加一个装饰器就能做到了:# Python 实用宝典
import functools
def catch_exceptions(cancel_on_failure=False):
def catch_exceptions_decorator(job_func):
@functools.wraps(job_func)
def wrapper(*args, **kwargs):
try:
return job_func(*args, **kwargs)
except:
import traceback
print(traceback.format_exc)
if cancel_on_failure:
return schedule.CancelJob
return wrapper
return catch_exceptions_decorator
@catch_exceptions(cancel_on_failure=True)
def bad_task:
return 1 / 0
schedule.every(5).minutes.do(bad_task)
这样,bad_task在执行时遇到的任何错误,都会被catch_exceptions 捕获,这点在保证调度任务正常运转的时候非常关键。
卓文科技打造物联网智慧微型融合断路器卓文科技打造物联网智慧微型融合断路器图卓文科技智慧电箱智慧微型融合断路器也叫智慧微断。生活中大家一般都叫智能断路器。当电路中电流产生过压或欠压状态自动断电,这样对使用中的电器产生保
卓文智慧用电监测平台,让用电管理更高效!卓文智慧用电监测平台,让用电管理更高效!卓文智慧用电监测平台是指通过物联网技术对引发电气火灾的主要因素(电流漏电流导线温度等)进行不间断的数据跟踪与统计分析,实时发现电气线路和用电
卓文科技再次携手华为,共同助力智能配电物联网建设卓文科技助力山东大学智慧用电改造项目顺利开展背景近年来,由于我国各大院校招生人数持续增长,学校人口高度密集,火灾事故频频发生。学校作为文化传承的重要基地,肩负着立德树人为国家培养现
普通家庭加装智能微型融合断路器让安全更加有保障普通家庭加装智能微型融合断路器让安全更加有保障随着城市化进程迅速发展,每年由于电气原因引发的火灾也不在少数,如何减少电气火灾的发生?除了向居民住户普及用电安全的常识以外,卓文科技运
卓文科技智能断路器实现智能化控制卓文科技智能断路器实现智能化控制对于现代需求的提升以及人们对安全用电意识的增强,传统断路器功能上逐渐无法满足日常生活中对用电安全的需求,卓文科技智能微型融合断路器在传统空开配电设备
卓文科技教您如何预防电气火灾?中国消防发布的相关火灾数据显示,2018年全国发生的火灾中,由电气原因引起的占总数的34。6,占比最大,而今年1至9月份全国火灾的起火原因,有32。8系用电引起,且45。1的较大火
智能断路器智能开关卓文科技智能断路器物联网空开智能开关智能断路器厂家卓文科技随着我国经济快速发展,电力设备需求增长迅速,与此同时,电气安全隐患量也一直居高不下。据2019年全国接报火灾数据统计,有52系电气
原来,电影中关于病毒肆虐的这些情节都不是真实根据2月25日关于猖獗病毒的报道,电影中虚构的故事和现实之间有多远?针对这个问题,秘鲁商报在2月22日给出了答案。这篇文章摘录如下显然,电影院和电视是人们了解科学家和医生工作的常见
2020年终于等到电影院开放了,你多久没去影院了7月20日上午,第23届上海国际电影节正式开票,上海的各家电影院也迎来了期盼已久的复工。早上950,天山电影院开始了上海在影院复工后的第一场放映。按照疫情防控的要求,每场上座率不超
骁龙710手机价格集体跳水,红米Note7压力很大荣耀9X上市之后,千元机市场的竞争已经进入了白热化的阶段,因为荣耀把麒麟810这个处理器直接干到了1399,这本来是骁龙710的价格区间。麒麟810到底有多强?之前笔者详细的介绍过
库克打响价格战,大内存iPhoneXR跳水2100,给iPhone11让路今年,大家仿佛把对iPhone的关注点都放在了iPhoneXR的身上,因为从今年年初开始,iPhoneXR的价格就开始循序渐进的滑坡,直到六月份时产生了一波猛降,马上就成为高端机市
售价79万元的它能够满足当下年轻人?比亚迪元Pro深度体验如果给你79万元预算,购买一辆纯电车的话,你会有怎样的选择?相信这个问题确实不太好回答,毕竟这样的预算在当下的纯电动车市场中能够找到满意答案的车型凤毛菱角。多数的车型因存在续航里程
今天支付宝集福正式开启!今天,2021年春节集五福活动正式开启,和往年相同,今年依然集齐五个福字即可合成五福红包,等到除夕夜开奖!支付宝在不断支招教大家使用新功能,目前除了原有的AR扫福写福字得福卡已经开
为什么要买新能源车?比亚迪唐dmi注意是我为什么会买,一切购买或推荐还是建立在个人需求上,所以我不是鼓吹新能源汽车的曰1。上牌方便,对于现在一线城市和超线城市,條牌的上牌和不限行不限导真的会方便很多。2。省钱,这个
唐Dmi电桩安装记(省钱大法)充电桩的安装历程真的是艰辛呀之前都是用小区的共享充电桩,电费1。7元,太贵了,现在有了自己的充电桩,家用电费0。49,实在!百公里十块钱以内接下来是充电桩安装流程1。电网申请装电表
我的新宠,帅气的大唐来了去年就打算买车,后面由于年底的工作耽误了一段时间,就推迟到了现在,之前有看过很多车型,但总有不如人意的地方,对于我这个选择困难症患者来说真的很难做出决定,全新唐还是朋友给推荐的,考
深度技术分享如何将win10壁纸设置高清的方法有不少深度技术的用户都不喜欢Windows10系统中,自带的桌面win10壁纸,在网上下载的不少壁纸又不够清晰,要如果设置高清壁纸呢?其实,这个问题真的很容易解决的,比如你可以搜索
比亚迪汉给了我十足的温暖自认生活条件还可以,属于比上不足比下有余,过的还算满足,买车的原因其实就两点,一个是挤地铁,时常会闻到不好闻的气味,另一个也是主要原因就是有了小孩,买个车子干什么都还是比较安全和方
朋友的比亚迪唐Dmi终于到家啦,最快速度的提车坐标上海,因为大牌贵且难拍,所以直接放弃要大牌的想法了,转战电动汽车,送牌照,完美!从年初就开始想买车了,看了现今市面上很火的各种电动汽车。第一个看的是小鹏p7,小鹏g3,p7的样
唐Dmi使用体验和驾驶感受用车感受1。动力这是大家最关心的,因为试驾过唐DM的车友都知道,那四秒多的加速,太猛了,秒天秒地秒空气。唐DMi的动力,完全不用担心!尤其是之前开燃油车的车友,就会感受到质的飞跃。
买车买的就是安全,刀片电池yyds今天,我们来聊聊爱车的安全问题。提起新能源车的安全,我们首先想到的肯定是电池的安全性,那么,我购买比亚迪汉ev的时候,选择原因肯定会有刀片电池了。毕竟汉ev是首款搭配刀片电池的新能
win10版本关闭防火墙提示关闭通知的解决方法很多深度技术的小伙伴在最新win10版本系统中关闭防火墙后,都会提示弹出关闭通知的问题,那么下面深度系统小编以win10正式版系统为例,来为大家分享具体的解决方法。方法如下1首先我