快捷導(dǎo)航

Python腳本實(shí)現(xiàn)定時(shí)任務(wù)的最佳方法

更新時(shí)間：2022年05月19日 15:41:17 作者：任識(shí)算法

我們?cè)谌粘９ぷ髦?常常會(huì)用到需要周期性執(zhí)行的任務(wù),下面這篇文章主要給大家介紹了關(guān)于Python腳本實(shí)現(xiàn)定時(shí)任務(wù)的最佳方法,文中通過(guò)實(shí)例代碼介紹的非常詳細(xì),需要的朋友可以參考下

前言

在日常工作中，常常需要周期性地執(zhí)行某些任務(wù)，常用的方式是采用 Linux 系統(tǒng)自帶的 crond 結(jié)合命令行實(shí)現(xiàn)，但最近卻遇到了一個(gè)讓人頭大的問(wèn)題。

問(wèn)題描述

一個(gè)包含cx_Oracle的python文件，直接在linux下使用python命令可以運(yùn)行，但是設(shè)置crontab定時(shí)任務(wù)會(huì)報(bào)錯(cuò)如下：

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

折騰半天，才找到問(wèn)題的根源，原來(lái)crontab進(jìn)程執(zhí)行shell腳本時(shí)不會(huì)自動(dòng)加載用戶目錄下的.profile文件，無(wú)法正常使用自定義的環(huán)境變量，所以就找不到oracle相關(guān)的庫(kù)，報(bào)錯(cuò)了。。。

解決方案

直接使用Python腳本實(shí)現(xiàn)定時(shí)任務(wù)，實(shí)現(xiàn)方式有很多：

利用while True: + sleep()實(shí)現(xiàn)定時(shí)任務(wù)
使用Timeloop庫(kù)運(yùn)行定時(shí)任務(wù)
利用threading.Timer實(shí)現(xiàn)定時(shí)任務(wù)
利用內(nèi)置模塊sched實(shí)現(xiàn)定時(shí)任務(wù)
利用調(diào)度模塊schedule實(shí)現(xiàn)定時(shí)任務(wù)
利用任務(wù)框架APScheduler實(shí)現(xiàn)定時(shí)任務(wù)
使用分布式消息系統(tǒng)Celery實(shí)現(xiàn)定時(shí)任務(wù)
使用數(shù)據(jù)流工具Apache Airflow實(shí)現(xiàn)定時(shí)任務(wù)

下面給大家展示一種任同學(xué)認(rèn)為最好用的實(shí)現(xiàn)方式：利用調(diào)度模塊schedule實(shí)現(xiàn)定時(shí)任務(wù)。

import schedule
import time

def job():
    print("I'm working...")

# 每十分鐘執(zhí)行任務(wù)
schedule.every(10).minutes.do(job)
# 每個(gè)小時(shí)執(zhí)行任務(wù)
schedule.every().hour.do(job)
# 每天的10:30執(zhí)行任務(wù)
schedule.every().day.at("10:30").do(job)
# 每個(gè)月執(zhí)行任務(wù)
schedule.every().monday.do(job)
# 每個(gè)星期三的13:15分執(zhí)行任務(wù)
schedule.every().wednesday.at("13:15").do(job)
# 每分鐘的第17秒執(zhí)行任務(wù)
schedule.every().minute.at(":17").do(job)

while True:
    schedule.run_pending()
    time.sleep(1)

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

如果你想只運(yùn)行一次任務(wù)的話，可以這么配：

import schedule
import time

def job_that_executes_once():
    # 此處編寫(xiě)的任務(wù)只會(huì)執(zhí)行一次...
    return schedule.CancelJob

schedule.every().day.at('22:30').do(job_that_executes_once)

while True:
    schedule.run_pending()
    time.sleep(1)

如果你有參數(shù)需要傳遞給作業(yè)去執(zhí)行，可以這么做：

import schedule

def greet(name):
    print('Hello', name)

# do() 將額外的參數(shù)傳遞給job函數(shù)
schedule.every(2).seconds.do(greet, name='Alice')
schedule.every(4).seconds.do(greet, name='Bob')

如果你需要讓某個(gè)定時(shí)任務(wù)到某個(gè)時(shí)間截止，可以這么做：

import schedule
from datetime import datetime, timedelta, time

def job():
    print('Boo')

# 每個(gè)小時(shí)運(yùn)行作業(yè)，18:30后停止
schedule.every(1).hours.until("18:30").do(job)

# 每個(gè)小時(shí)運(yùn)行作業(yè)，2030-01-01 18:33 today
schedule.every(1).hours.until("2030-01-01 18:33").do(job)

# 每個(gè)小時(shí)運(yùn)行作業(yè)，8個(gè)小時(shí)后停止
schedule.every(1).hours.until(timedelta(hours=8)).do(job)

# 每個(gè)小時(shí)運(yùn)行作業(yè)，11:32:42后停止
schedule.every(1).hours.until(time(11, 33, 42)).do(job)

# 每個(gè)小時(shí)運(yùn)行作業(yè)，2020-5-17 11:36:20后停止
schedule.every(1).hours.until(datetime(2020, 5, 17, 11, 36, 20)).do(job)

怎么樣？是不是特好用!

總結(jié)

到此這篇關(guān)于Python腳本實(shí)現(xiàn)定時(shí)任務(wù)的最佳方法的文章就介紹到這了,更多相關(guān)Python實(shí)現(xiàn)定時(shí)任務(wù)內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: