一. 前言

一個(gè)網(wǎng)頁會(huì)有很多數(shù)據(jù)是不需要經(jīng)常變動(dòng)的，比如說首頁，變動(dòng)頻率低而訪問量大，我們可以把它靜態(tài)化，這樣就不需要每次有請(qǐng)求都要查詢數(shù)據(jù)庫再返回，可以減少服務(wù)器壓力
我們可以使用Django的模板渲染功能完成頁面渲染

二. APSchedule/django-apschedule簡(jiǎn)介

APScheduler的全稱是Advanced Python Scheduler。它是一個(gè)輕量級(jí)的 Python 定時(shí)任務(wù)調(diào)度框架。
APScheduler 支持三種調(diào)度任務(wù)：

• 固定時(shí)間間隔
• 固定時(shí)間點(diǎn)（日期）
• Linux 下的 Crontab命令。同時(shí)，它還支持異步執(zhí)行、后臺(tái)執(zhí)行調(diào)度任務(wù)。

特點(diǎn)

1）可以動(dòng)態(tài)添加任務(wù)
2）不依賴Linux的crontab系統(tǒng)定時(shí)
3）可以對(duì)添加的定時(shí)任務(wù)做持久保存

之前已經(jīng)介紹過APScheduler的使用，下面介紹的是django-apscheduler的使用

三. 【django-apscheduler】使用

1.安裝APScheduler

 pip install django-apscheduler

2. 使用Django_apscheduler步驟

1.創(chuàng)建app

python manage.py startapp test

2. 注冊(cè)使用

在settings.py中注冊(cè)django-apscheduler和test

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'django_apscheduler',
    'apps.test'	# 注意，我這里是在外層包了一個(gè)apps的文件夾，test為業(yè)務(wù)模塊
]

3.在test文件夾中新建urls.py

在子模塊apps/test/urls.py中添加如下代碼

from django.urls import path
from apps.test import views

urlpatterns = [
]

4. 在項(xiàng)目總路由urls.py中添加test.urls

打開djangoproject中的urls.py,輸入如下代碼

from django.contrib import admin
from django.urls import path, include

urlpatterns = [
    path('admin/', admin.site.urls),
    path('', include('apps.test.urls')),
]

5. 執(zhí)行遷移

python manage.py makemigrations
python manage.py migrate

沒有其他表結(jié)構(gòu)的話不必運(yùn)行 python manage.py makemigrations
會(huì)創(chuàng)建兩張表：

• django_apscheduler_djangojob
• django_apscheduler_djangojobexecution

通過進(jìn)入后臺(tái)管理能方便管理定時(shí)任務(wù)。

django_apscheduler_djangojob 表保存注冊(cè)的任務(wù)以及下次執(zhí)行的時(shí)間

這里注意 status為executed是執(zhí)行，missed 則是表示撞車的場(chǎng)景, 為避免這種場(chǎng)景需要在 周期的長(zhǎng)度以及是否進(jìn)行強(qiáng)制結(jié)束進(jìn)行選擇

6. 在test子應(yīng)用中的urls.py中輸入下面的代碼

from django.shortcuts import render

# Create your views here.
from apscheduler.schedulers.background import BackgroundScheduler
from django_apscheduler.jobstores import DjangoJobStore, register_events, register_job,DjangoResultStoreMixin

# 實(shí)例化調(diào)度器
scheduler = BackgroundScheduler()
# 開啟定時(shí)工作
# 調(diào)度器使用DjangoJobStore()
scheduler.add_jobstore(DjangoJobStore(), "default")


# 設(shè)置定時(shí)任務(wù)，選擇方式為interval，時(shí)間間隔為10s
# 另一種方式為每天固定時(shí)間執(zhí)行任務(wù)，對(duì)應(yīng)代碼為：
# @register_job(scheduler, 'cron', day_of_week='mon-fri', hour='9', minute='30', second='10',id='task_time')
@register_job(scheduler, "interval", seconds=10,id='test_job', replace_existing=True)  # replace_existing=解決第二次啟動(dòng)失敗的問題
def my_job():
    # 這里寫你要執(zhí)行的任務(wù)
    pass

# register_events(scheduler)    最新的django_apscheduler已經(jīng)不需要這一步
scheduler.start()

最好為job加上id ，不加也可以

注意： 需要加上replace_existing=True 否則會(huì)報(bào)以下錯(cuò)誤，即id重複

raise ConflictingIdError(job.id)
apscheduler.jobstores.base.ConflictingIdError: 'Job identifier (index_html) conflicts with an existing job'

提示：也可以不寫在Django工程目錄下的urls.py文件中（主urls.py）或者子應(yīng)用的urls.py 中，百度說可以寫在view.py 中或其他隨便哪個(gè)文件中，但是我沒試成功過

7. 運(yùn)行django項(xiàng)目

python manage.py runserver 8000

常見的三種調(diào)度參數(shù)

• date：希望在某個(gè)特定時(shí)間僅運(yùn)行一次作業(yè)時(shí)使用
• interval：要以固定的時(shí)間間隔運(yùn)行作業(yè)時(shí)使用
• cron：以crontab的方式運(yùn)行定時(shí)任務(wù)

3. 示例

from apscheduler.schedulers.background import BackgroundScheduler
from django_apscheduler.jobstores import DjangoJobStore, register_events, register_job

scheduler = BackgroundScheduler()
scheduler.add_jobstore(DjangoJobStore(), "default")


# 時(shí)間間隔3秒鐘打印一次當(dāng)前的時(shí)間
@register_job(scheduler, "interval", seconds=3, id='test_job')
def test_job():
    print("我是apscheduler任務(wù)")
# per-execution monitoring, call register_events on your scheduler
register_events(scheduler)
scheduler.start()
print("Scheduler started!")

運(yùn)行結(jié)果

Scheduler started!
我是apscheduler任務(wù)
我是apscheduler任務(wù)
...

APScheduler中兩種調(diào)度器的區(qū)別及使用過程中要注意的問題

APScheduler中有很多種不同類型的調(diào)度器，BlockingScheduler與BackgroundScheduler是其中最常用的兩種調(diào)度器。區(qū)別主要在于BlockingScheduler會(huì)阻塞主線程的運(yùn)行，而BackgroundScheduler不會(huì)阻塞。所以，我們?cè)诓煌那闆r下，選擇不同的調(diào)度器：

• BlockingScheduler: 調(diào)用start函數(shù)后會(huì)阻塞當(dāng)前線程。當(dāng)調(diào)度器是你應(yīng)用中唯一要運(yùn)行的東西時(shí)（如上例）使用
• BackgroundScheduler: 調(diào)用start后主線程不會(huì)阻塞。當(dāng)你不運(yùn)行任何其他框架時(shí)使用，并希望調(diào)度器在你應(yīng)用的后臺(tái)執(zhí)行。

BlockingScheduler的示例

from apscheduler.schedulers.blocking import BlockingScheduler
import time

def job():
    print('job 3s')


if __name__=='__main__':

    sched = BlockingScheduler(timezone='MST')
    sched.add_job(job, 'interval', id='3_second_job', seconds=3)
    sched.start()

    while(True):
        print('main 1s')
        time.sleep(1)

運(yùn)行結(jié)果

job 3s
job 3s
job 3s
...

可見，BlockingScheduler調(diào)用start函數(shù)后會(huì)阻塞當(dāng)前線程，導(dǎo)致主程序中while循環(huán)不會(huì)被執(zhí)行到。

BackgroundScheduler的例子

from apscheduler.schedulers.background import BackgroundScheduler
import time

def job():
    print('job 3s')


if __name__=='__main__':

    sched = BackgroundScheduler(timezone='MST')
    sched.add_job(job, 'interval', id='3_second_job', seconds=3)
    sched.start()

    while(True):
        print('main 1s')
        time.sleep(1)

運(yùn)行結(jié)果

main 1s
main 1s
main 1s
job 3s
main 1s
main 1s
main 1s
job 3s
...

可見，BackgroundScheduler調(diào)用start函數(shù)后并不會(huì)阻塞當(dāng)前線程，所以可以繼續(xù)執(zhí)行主程序中while循環(huán)的邏輯。

通過這個(gè)輸出，我們也可以發(fā)現(xiàn)，調(diào)用start函數(shù)后，job()并不會(huì)立即開始執(zhí)行。而是等待3s后，才會(huì)被調(diào)度執(zhí)行。

如何讓job在start()后就開始運(yùn)行

其實(shí)APScheduler并沒有提供很好的方法來解決這個(gè)問題，但有一種最簡(jiǎn)單的方式，就是在調(diào)度器start之前，就運(yùn)行一次job()，如下

from apscheduler.schedulers.background import BackgroundScheduler
import time

def job():
    print('job 3s')


if __name__=='__main__':
    job()
    sched = BackgroundScheduler(timezone='MST')
    sched.add_job(job, 'interval', id='3_second_job', seconds=3)
    sched.start()

    while(True):
        print('main 1s')
        time.sleep(1)

這樣雖然沒有絕對(duì)做到讓job在start()后就開始運(yùn)行，但也能做到不等待調(diào)度，而是剛開始就運(yùn)行job

如果job執(zhí)行時(shí)間過長(zhǎng)會(huì)怎么樣

如果執(zhí)行job()的時(shí)間需要5s，但調(diào)度器配置為每隔3s就調(diào)用一下job()，會(huì)發(fā)生什么情況呢？我們寫了如下例子

from apscheduler.schedulers.background import BackgroundScheduler
import time

def job():
    print('job 3s')
    time.sleep(5)

if __name__=='__main__':

    sched = BackgroundScheduler(timezone='MST')
    sched.add_job(job, 'interval', id='3_second_job', seconds=3)
    sched.start()

    while(True):
        print('main 1s')
        time.sleep(1)

可見，3s時(shí)間到達(dá)后，并不會(huì)“重新啟動(dòng)一個(gè)job線程”，而是會(huì)跳過該次調(diào)度，等到下一個(gè)周期（再等待3s），又重新調(diào)度job()。

為了能讓多個(gè)job()同時(shí)運(yùn)行，我們也可以配置調(diào)度器的參數(shù)max_instances，如下例，我們?cè)试S2個(gè)job()同時(shí)運(yùn)行

from apscheduler.schedulers.background import BackgroundScheduler
import time

def job():
    print('job 3s')
    time.sleep(5)

if __name__=='__main__':
    job_defaults = { 'max_instances': 2 }
    sched = BackgroundScheduler(timezone='MST', job_defaults=job_defaults)
    sched.add_job(job, 'interval', id='3_second_job', seconds=3)
    sched.start()

    while(True):
        print('main 1s')
        time.sleep(1)

我們可看到如下運(yùn)行結(jié)果

每個(gè)job是怎么被調(diào)度的

通過上面的例子，我們發(fā)現(xiàn)，調(diào)度器是定時(shí)調(diào)度job()函數(shù)，來實(shí)現(xiàn)調(diào)度的。
那job()函數(shù)會(huì)被以進(jìn)程的方式調(diào)度運(yùn)行，還是以線程來運(yùn)行呢？
為了弄清這個(gè)問題，我們寫了如下程序：

from apscheduler.schedulers.background import BackgroundScheduler
import time,os,threading

def job():
    print('job thread_id-{0}, process_id-{1}'.format(threading.get_ident(), os.getpid()))
    time.sleep(50)

if __name__=='__main__':
    job_defaults = { 'max_instances': 20 }
    sched = BackgroundScheduler(timezone='MST', job_defaults=job_defaults)
    sched.add_job(job, 'interval', id='3_second_job', seconds=3)
    sched.start()

    while(True):
        print('main 1s')
        time.sleep(1)

我們可看到如下運(yùn)行結(jié)果

可見，每個(gè)job()的進(jìn)程ID都相同，但線程ID不同。所以，job()最終是以線程的方式被調(diào)度執(zhí)行

BlockingScheduler定時(shí)任務(wù)及其他方式的實(shí)現(xiàn)

#BlockingScheduler定時(shí)任務(wù)
from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
首先看看周一到周五定時(shí)執(zhí)行任務(wù)
# 輸出時(shí)間
def job():
  print(datetime.now().strtime("%Y-%m-%d %H:%M:%S"))
# BlockingScheduler
scheduler = BlockingScheduler()
scheduler.add_job(job, "cron"， day_of_week="1-5", hour=6, minute=30)
schduler.start()
 
 
scheduler.add_job(job, 'cron', hour=1, minute=5)
hour =19 , minute =23  這里表示每天的19：23 分執(zhí)行任務(wù)
hour ='19', minute ='23'  這里可以填寫數(shù)字，也可以填寫字符串
hour ='19-21', minute= '23'  表示 19:23、 20:23、 21:23 各執(zhí)行一次任務(wù)
 
#每300秒執(zhí)行一次
scheduler .add_job(job, 'interval', seconds=300)
 
#在1月,3月,5月,7-9月，每天的下午2點(diǎn)，每一分鐘執(zhí)行一次任務(wù)
scheduler .add_job(func=job, trigger='cron', month='1,3,5,7-9', day='*', hour='14', minute='*')
 
# 當(dāng)前任務(wù)會(huì)在 6、7、8、11、12 月的第三個(gè)周五的 0、1、2、3 點(diǎn)執(zhí)行
scheduler .add_job(job, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')
 
#從開始時(shí)間到結(jié)束時(shí)間，每隔倆小時(shí)運(yùn)行一次
scheduler .add_job(job, 'interval', hours=2, start_date='2018-01-10 09:30:00', end_date='2018-06-15 11:00:00')
 
#自制定時(shí)器
 from datetime import datetime
 import time
 # 每n秒執(zhí)行一次
 def timer(n):
   while True:
     print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
     time.sleep(n)
 
timer(5)

BackgroundScheduler定時(shí)任務(wù)及其他方式的實(shí)現(xiàn)

啟動(dòng)異步定時(shí)任務(wù)
 import time
 from apscheduler.schedulers.background import BackgroundScheduler
 from django_apscheduler.jobstores import DjangoJobStore, register_events, register_job
 try: 
    # 實(shí)例化調(diào)度器
    scheduler = BackgroundScheduler()
    # 調(diào)度器使用DjangoJobStore()
    scheduler.add_jobstore(DjangoJobStore(), "default")
    # 'cron'方式循環(huán)，周一到周五，每天9:30:10執(zhí)行,id為工作ID作為標(biāo)記
    # ('scheduler',"interval", seconds=1) #用interval方式循環(huán)，每一秒執(zhí)行一次
    @register_job(scheduler, 'cron', day_of_week='mon-fri', hour='9', minute='30', second='10',id='task_time')
    def test_job():
      t_now = time.localtime()
      print(t_now)
  
   # 監(jiān)控任務(wù)
   register_events(scheduler)
   # 調(diào)度器開始
   scheduler.start()
except Exception as e:
  print(e)
  # 報(bào)錯(cuò)則調(diào)度器停止執(zhí)行
  scheduler.shutdown()

參考鏈接：https://www.cnblogs.com/guojie-guojie/p/16330165.html
以上就是python- 定時(shí)任務(wù)框架【APScheduler】基本使用介紹，希望對(duì)你有所幫助！文章來源地址http://www.zghlxwxcb.cn/news/detail-525888.html

到了這里，關(guān)于Django - 定時(shí)任務(wù)框架【django-apscheduler】基本使用詳解(二)的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！