Python脚本实现定时任务的最佳方法_Python

Python脚本实现定时任务的最佳方法

2023-02-10 10:53任识算法 Python

我们在日常工作中,常常会用到需要周期性执行的任务,下面这篇文章主要给大家介绍了关于Python脚本实现定时任务的最佳方法,文中通过实例代码介绍的非常详细,需要的朋友可以参考下

前言

在日常工作中，常常需要周期性地执行某些任务，常用的方式是采用 Linux 系统自带的 crond 结合命令行实现，但最近却遇到了一个让人头大的问题。

问题描述

一个包含cx_Oracle的python文件，直接在linux下使用python命令可以运行，但是设置crontab定时任务会报错如下：

cx_Oracle.DatabaseError:

DPI-1047: 64-bit Oracle Client library cannot be loaded:

"libclntsh.so: cannot open shared object file: No such file or directory".

See https://oracle.github.io/odpi/doc/installation.html#linux for help

折腾半天，才找到问题的根源，原来crontab进程执行shell脚本时不会自动加载用户目录下的.profile文件，无法正常使用自定义的环境变量，所以就找不到oracle相关的库，报错了。。。

解决方案

直接使用Python脚本实现定时任务，实现方式有很多：

利用while True: + sleep()实现定时任务
使用Timeloop库运行定时任务
利用threading.Timer实现定时任务
利用内置模块sched实现定时任务
利用调度模块schedule实现定时任务
利用任务框架APScheduler实现定时任务
使用分布式消息系统Celery实现定时任务
使用数据流工具Apache Airflow实现定时任务

下面给大家展示一种任同学认为最好用的实现方式：利用调度模块schedule实现定时任务。

				?

									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)

可以看到，从月到秒的配置，上面的例子都覆盖到了。

如果你想只运行一次任务的话，可以这么配：

				?

									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)

如果你有参数需要传递给作业去执行，可以这么做：

				?

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

如果你需要让某个定时任务到某个时间截止，可以这么做：

				?

									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)