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

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

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

前言

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

問題描述

一個包含cx_Oracle的python文件，直接在linux下使用python命令可以運行，但是設(shè)置crontab定時任務(wù)會報錯如下：

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進(jìn)程執(zhí)行shell腳本時不會自動加載用戶目錄下的.profile文件，無法正常使用自定義的環(huán)境變量，所以就找不到oracle相關(guān)的庫，報錯了。。。

解決方案

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

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

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

import schedule
import time

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

# 每十分鐘執(zhí)行任務(wù)
schedule.every(10).minutes.do(job)
# 每個小時執(zhí)行任務(wù)
schedule.every().hour.do(job)
# 每天的10:30執(zhí)行任務(wù)
schedule.every().day.at("10:30").do(job)
# 每個月執(zhí)行任務(wù)
schedule.every().monday.do(job)
# 每個星期三的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)

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

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

import schedule
import time

def job_that_executes_once():
    # 此處編寫的任務(wù)只會執(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')

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

import schedule
from datetime import datetime, timedelta, time

def job():
    print('Boo')

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

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

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

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

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

怎么樣？是不是特好用!