Python 周期任务神器,太实用了

前言

如果你想在Linux服务器上周期性地执行某个 Python 脚本,最出名的选择应该是 Crontab 脚本,但是 Crontab 具有以下缺点:

1.不方便执行秒级的任务。

2.当需要执行的定时任务有上百个的时候,Crontab的管理就会特别不方便。

另外一个选择是 Celery,但是 Celery 的配置比较麻烦,如果你只是需要一个轻量级的调度工具,Celery 不会是一个好选择。

在你想要使用一个轻量级的任务调度工具,而且希望它尽量简单、容易使用、不需要外部依赖,最好能够容纳 Crontab 的所有基

本功能,那么 Schedule 模块是你的不二之选。

使用它来调度任务可能只需要几行代码就能感觉到:

[En]

Using it to schedule tasks may take only a few lines of code to feel:

python源码/素材/插件:903971231####
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.准备

请选择以下任何一种方式输入命令以安装依赖项:

[En]

Please choose any of the following ways to enter the command to install dependencies:

Windows 环境 打开 Cmd (开始-运行-CMD)。
MacOS 环境 打开 Terminal (command+空格输入Terminal)。
如果你用的是 VSCode编辑器 或 Pycharm,可以直接使用界面下方的Terminal.

pip install schedule

2.基本使用

本文开头提到了最基本的用法。以下是更多调度任务的示例:

[En]

The most basic use has been mentioned at the beginning of the article. Here are more examples of scheduling tasks:

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)

如您所见,上面的示例涵盖了每个月到第二个月的配置。但如果您只想运行一次任务,您可以这样匹配它:

[En]

As you can see, the above examples cover the configuration from month to second. But if you want to run a task only once, you can match it like this:

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)

参数传递

如果您有要传递给作业执行的参数,则只需执行以下操作:

[En]

If you have parameters to pass to the job to execute, you just need to do this:

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')

获取目前所有的作业

如果您想获得所有当前任务:

[En]

If you want to get all the current assignments:

Python 实用宝典
import schedule

def hello():
    print('Hello world')

schedule.every().second.do(hello)

all_jobs = schedule.get_jobs()

Python 周期任务神器,太实用了

取消所有作业

如果触发了某种机制,您需要立即从当前程序中清除所有作业:

[En]

If some mechanism is triggered, you need to immediately clear all jobs from the current program:

Python 实用宝典
import schedule

def greet(name):
    print('Hello {}'.format(name))

schedule.every().second.do(greet)

schedule.clear()

标签功能

在设置作业时,为了以后管理作业,您可以对作业进行标记,以便通过标记筛选获得作业或取消作业。

[En]

When setting up a job, in order to manage the job later, you can tag the job so that you can get the job or cancel the job through tag filtering.

Python 实用宝典
python插件/源码/素材加Q群:903971231####
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')

设定作业截止时间

如果您需要在某个时间进行到期的作业,您可以这样做:

[En]

If you need to make an assignment due at a certain time, you can do it this way:

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)

在截止日期之后,该作业将不会运行。

[En]

After the deadline, the job will not run.

立即运行所有作业,而不考虑其计划

[En]

Run all jobs immediately, regardless of their scheduling

如果某个机制触发了,你需要立即运行所有作业,可以调用 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秒<details><summary>*<font color='gray'>[En]</font>*</summary>*<font color='gray'>Run all jobs immediately with an interval of 10 seconds per job</font>*</details>
schedule.run_all(delay_seconds=10)

3.高级使用

装饰器安排作业

如果你觉得布置作业的形式太繁琐,你也可以使用装饰者模式:

[En]

If you think the form of setting homework is too verbose, you can also use decorator mode:

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 按顺序执行所有作业。其背后的原因是,很难找到让每个人都高兴的并行执行模型。

但是,您可以以多个线程的形式运行每个作业,以解决此限制:

[En]

However, you can run each job in the form of multiple threads to address this limitation:

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 不会自动捕捉异常,它遇到异常会直接抛出,这会导致一个严重的问题:后续所有的作业都会被中断执行,因此我们需要捕捉到这些异常。

您可以手动捕获,但一些意外情况需要程序自动捕获,这可以通过添加一个修饰器来完成:

[En]

You can capture it manually, but some unexpected situations need to be captured automatically by the program, which can be done by adding a decorator:

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 周期任务神器,太实用了

Original: https://www.cnblogs.com/1234567FENG/p/16401982.html
Author: 蚂蚁爱Python
Title: Python 周期任务神器,太实用了

原创文章受到原创版权保护。转载请注明出处:https://www.johngo689.com/499257/

转载文章受原作者版权保护。转载请注明原作者出处!

(0)

大家都在看

亲爱的 Coder【最近整理,可免费获取】👉 最新必读书单  | 👏 面试题下载  | 🌎 免费的AI知识星球