django 定時任務第三方庫apscheduler

1、apscheduler未安裝的需要自己安裝，安裝命令如下：

pip install apscheduler

　　

2、apscheduler庫有很多不同類型的調度器，其他我也不懂，只知道BlockingScheduler與BackgroundScheduler，這是比較常用的兩種
　　區別主要在於BlockingScheduler會阻塞主線程的運行，而BackgroundScheduler不會阻塞。所以，我們在不同的情況下，選擇不同的調度器：
　　BlockingScheduler: 調用start函數后會阻塞當前線程。當調度器是你應用中唯一要運行的東西時使用(下面寫的就是這種調度)
　　BackgroundScheduler: 調用start后主線程不會阻塞。當你不運行任何其他框架時使用，並希望調度器在你應用的后台執行
　　用人話說就是：用BlockingScheduler創建的類，調用start()函數開始運行后，會阻塞（相當於停止）當前已在運行的線程，只執行當前的定時任務，運行到你主動關掉為止
　　而BackgroundScheduler創建的類，調用start后主線程不會阻塞，會繼續運行下一行代碼，如果后面沒有死循環什么的，訂的時間又比較長，會出現定時還沒生效，代碼已經運行結束的情況
　　參考文章：https://www.cnblogs.com/will-wu/p/14721592.html

 

3、apscheduler的觸發器
　　apscheduler的觸發器有三種：date（日期觸發），interval（固定間隔觸發），cron（周期觸發）

　　　　date表示具體的一次性任務，interval表示循環任務，cron表示定時任務　　

　　3.1、日期觸發：date，特定的時間點觸發，最基本的一種調度，作業只會執行一次
　　　　run_date(datetime or str) 任務運行的日期或者時間
　　　　timezone(datetime.tzinfo or str) 指定時區

# 導入BlockingScheduler調度器
from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime

# 封裝函數待用
def print_hours_now():
    print('小時：%s' % datetime.now().strftime("%Y-%m-%d %H:%M:%S"))

# 創建一個對象
timing_type = BlockingScheduler()

# run_date時間到了之后，開始執行print_hours_now
timing_type.add_job(print_hours_now,'date',run_date='2021-04-30 16:13:00')
# 或這樣寫
# timing_type.add_job(func=print_hours_now,trigger='date',run_date='2021-04-30 16:13:00')

# 立即執行
timing_type.add_job(print_hours_now,'date')

# 等待觸發時間到達，開始執行指定任務
timing_type.start()

　　

3.2、interval：固定間隔觸發
　　　　year：每隔X年執行一次
　　　　weeks：每隔X周執行一次 weeks=X，范圍1-12
　　　　days：每隔X天執行一次，范圍1-31
　　　　hours: 每隔幾小時執行一次 | hours=0，范圍0-23
　　　　minutes: 每隔幾分執行一次 | minutes=0，范圍0-59
　　　　seconds: 每隔幾秒執行一次 | seconds=0，范圍0-59
　　　　start_date: 最早執行時間 | start_date=None，范圍是年月日的組合字符串
　　　　end_date: 最晚執行時間 | end_date=None，范圍是年月日的組合字符串
　　　　timezone: 執行時間區間 | timezone=None，范圍是年月日的組合字符串

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime

def print_minutes_now():
    print('打印時間：%s' % datetime.now().strftime("%Y-%m-%d %H:%M:%S"))

timing_type = BlockingScheduler()
# 在start_date到end_date時間段內，每隔5s執行一次print_minutes_now；其中，func=這些可以省略，start_date和end_date可以單獨使用，
# timing_type.add_job(func=print_minutes_now,trigger='interval',seconds=5,start_date='2021-05-04 11:40',end_date='2021-05-04 11:41:00')
# 在end_date之前，每隔0.2小時運行一次print_minutes_now
timing_type.add_job(func=print_minutes_now,trigger='interval',minutes=2,end_date='2021-05-08 11:41:00')
# 每隔1小時執行一次
timing_type.add_job(func=print_minutes_now,trigger='interval',hours=1)
# 啟動調度器
timing_type.start()

　　

3.3、cron：周期觸發，在特定時間周期性地觸發
　　　　cron參數：
　　　　　　year(int or str) 年，4位數字
　　　　　　month(int or str) 月（范圍1-12）
　　　　　　day(int or str) 日（范圍1-31）
　　　　　　week(int or str) 周（范圍1-53）
　　　　　　day_of_week(int or str) 周內第幾天或者星期幾（范圍0-6或者mon,tue,wed,thu,fri,stat,sun）
　　　　　　hour(int or str) 時（0-23）
　　　　　　minute(int or str) 分（0-59）
　　　　　　second(int or str) 秒（0-59）
　　　　　　start_date(datetime or str) 最早開始日期（含）
　　　　　　end_date(datetime or str) 最晚結束日期（含）
　　　　　　timezone(datetime.tzinfo or str) 指定時區

　　　　表達式：
　　　　　　* 所有 通配符。例：minutes=*即每分鍾觸發
　　　　　　*/a 所有 可被a整除的通配符。
　　　　　　a-b 所有 范圍a-b觸發
　　　　　　a-b/c 所有 范圍a-b，且可被c整除時觸發
　　　　　　xth y 日 第幾個星期幾觸發。x為第幾個，y為星期幾
　　　　　　last x 日 一個月中，最后個星期幾觸發
　　　　　　last 日 一個月最后一天觸發
　　　　　　x,y,z 所有 組合表達式，可以組合確定值或上方的表達式

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime

def print_time_now():
    print('打印當前時間：',datetime.now().strftime("%Y-%m-%d %H:%M:%S"))

timing_type = BlockingScheduler()
# 每個月每天的17:36,打印當前時間，*代表0-59
# timing_type.add_job(print_time_now,'cron',month='*',day='*',hour=17,minute=36,second='*')
# 每天的每小時16分的零秒開始到59秒執行print_time_now（年月日時不寫的時候會默認為所有，秒不寫則默認為0）
timing_type.add_job(func=print_time_now, trigger='cron',minute='16',second='0-59')
# 周一到周五6:30執行timing_type.add_job(print_time_now, "cron",day_of_week = "1-5", hour = 6, minute = 30)
# 啟動調度器 timing_type.start()

　　

示例：

#封裝郵件發送方法
from django.core.mail import send_mail


def send_email(email_user):
    send_mail(
        subject='這里是郵件標題',
        message='這里是郵件內容',
        from_email='[e-mail]',  # 發件人
        recipient_list=[email_user],  # 收件人
        #收件人可以直接寫，也可以從setting.py中配置中導入
        fail_silently=False
    )
    print("郵件發送成功")
    return "郵件已發送"


#定時任務
from apscheduler.schedulers.background import BackgroundScheduler

#待定時執行的函數
def update_days():
    all_borrow_book = Borrow.objects.filter(borrow_status=2)
    now = datetime.date.today()
    # all_bookname = all_borrow_book.book_id
    for book in all_borrow_book:
        id = book.id
        user_id = book.user_id
        book_id = book.book_id
        print("userid:",user_id)
        email_user = User.objects.get(username=user_id).email
        lib_book = Book.objects.get(book_number=book_id)
        update_book = Borrow.objects.get(id=id)
        return_date = book.return_date
        update_days = return_date.__sub__(now).days
        update_book.surplus_days = update_days
        if update_days < 0:
            update_book.borrow_status = -1
            update_book.save()
            lib_book.book_status = -1
            lib_book.save()
            send_email(email_user)
        else:
            update_book.save()
        print("surplus_days: ",update_days)


#開啟定時工作
try:
    # 實例化調度器
    scheduler = BackgroundScheduler()
    # scheduler.add_job(my_job, 'interval', minutes=10)
    scheduler.add_job(update_days,'cron',month='*',day='*',hour=19,minute=38,second=00)
    scheduler.start()
except Exception as e:
    print(e)

　　

https://apscheduler.readthedocs.io/en/3.x/userguide.html#code-examples

https://www.cnblogs.com/will-wu/p/14731728.html