简单强大的Python库!Schedule—实用的周期任务调度工具转载

原创
小哥 3年前 (2022-10-16) 阅读数 58 #大杂烩

如果你想工作如果你想工作如果你想工作Linux服务器周期性地执行某一项,服务器周期性地执行某项,服务器周期性地执行某一项 Python 剧本,最著名的选择是剧本,最著名的选择应该是 Crontab 脚本,但是脚本,但是脚本,但是 Crontab 有以下缺点。有以下缺点。有以下几个缺点。

1.执行不方便执行不方便执行不容易执行 **任务秒数任务** 。

2.当有数百个定时任务要执行时Crontab的 **管理会特别不方便管理会特别不方便** 。

另一个选项是另一个选项 Celery,但是 Celery 配置起来有点麻烦,如果您只需要一个轻量级的调度工具,Celery 那可不是个好选择。不是一个好的选择。这不是一个好的选择。

如果您想要使用轻量级任务调度工具,并且希望它尽可能简单易用,没有外部依赖,并且最好能够适应 Crontab 的所有基本功能 Schedule 该模块是您正确的选择。模块是您的正确选择。模块由您选择。

使用它来调度任务可能只需要几行代码就可以体验一下。

# Python 实用宝典
import schedule
import time

def job():
    print("Im 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("Im working...")

# 每10分钟执行任务每10分钟执行任务每10分钟执行任务每10分钟执行任务
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("Im running on thread %s" % threading.current_thread())
def job2():
    print("Im running on thread %s" % threading.current_thread())
def job3():
    print("Im 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 捕获,这是保持调度任务正常运行的关键。

这是我们文章的结尾,如果你喜欢今天的Python 实用教程,敬请关注实用教程,敬请关注Python一个实用的宝藏。实用的宝藏。实用的宝藏。

**关于我们星球俱乐部的最后一句话。关于我们的星球俱乐部,最后一句话:。最后,谈一谈我们的星球俱乐部。**

如果你觉得自学不堪重负,如果你一个人学习很无聊,就像在天堂看书一样,那么材料太多了,不知道该选哪本书看。想学习遇到问题有人咨询,想通过有趣的案例加速学习,欢迎加入我们的团队!

一年88人民币现贬值人民币现贬值美元现贬值现贬值12元,

大约每个月,几乎每个月,只要大约每个月,少到每个月,只需要6物有所值非常物有所值物有所值

推荐阅读:
入门: 最完整的零基学习最全面的零基学习最完整的零基学习Python的问题  | 从零开始学习从零基础学习从零基础学习8个月的Python  | 实战项目 |学Python这是捷径,这是捷径,这是捷径
干货:爬行豆瓣短评,电影《后来的我们》 | 38年NBA最佳球员分析最佳球员分析 |   从万众期待到口碑惨败!唐探3令人失望  | 笑新伊田图龙记笑新伊田图龙记笑新伊田图龙记 | 谜语之王回答灯谜之王灯谜之王谜语之王 |用Python人山人海素描图人山人海素描图人山人海 Dishonor太火了,我用机器学习做了一个迷你推荐系统电影
趣味:弹球游戏  | 九宫格  | 漂亮的花 | 两百行Python日常酷跑游戏日常酷跑游戏日常酷跑游戏!
AI: 会写诗的机器人会写诗的机器人会写诗的机器人 | 给图片上色给图片上色给图片上色 | 预测收入 | 《耻辱》太火了,我用机器学习做了一部迷你推荐系统电影
小工具: Pdf转Word易于修复表单和水印!易于处理的表单和水印!易于修复表单和水印!简单的表格和水印! | 一键把html将页面另存为网页另存为网页另存为pdf!|  再见PDF提款费!提款费!提款费!提款费用! | 用90构建最强大的代码行构建最强大的代码行构建最强大的代码行PDF转换器,word、PPT、excel、markdown、html一键转换 | 制作一个固定的低成本机票提醒!制作一张别针的低价机票提醒! |60代码行做了一个语音墙纸切换,天天见女士!

年度弹出文案年度弹出文案年度爆炸性文案

  • 1). 卧槽!Pdf转Word用Python轻松搞定 !

  • 2).学Python闻起来好香!我用100一行代码做了一个网站,帮助人们做了一行代码,做了一个网站,帮助了人们做了一行代码,帮助了人们PS旅行图片赚鸡腿吃旅行图片赚鸡腿

  • 3).第一次播放量过亿,火爆全网,我分析了《波妹》,发现了这些秘密

  • 4). 80一行行代码!使用Python让救济金做正确的事做做的人做好事的人A梦分身

  • 5).你必须掌握的东西你必须掌握20个python代码,简短而紧凑,永无止境的有用代码,简短而甜蜜,永无止境的有用的代码,简短而紧凑,永无止境的使用代码,简短而甜蜜,永无止境的用途

  • 6). 30个Python古怪技能集古怪小贴士收藏古怪技能集

  • 7). 我总结的80《菜鸟学习专页》《菜鸟学习专页》《菜鸟学习》Python精选干货.pdf》,都是干货

  • 8). 再见Python!我要学Go了!2500词深度分析词深度分析词深度分析 !

  • 9).发现了一只舔狗的福利!这Python爬虫神器太酷了,不能自动下载女孩的照片

点击阅读原文点击查看点击点击阅读点击阅读原文点击查看200个Python案例!

版权声明

所有资源都来源于爬虫采集,如有侵权请联系我们,我们将立即删除

热门