django使用定時(shí)任務(wù)django_apscheduler的實(shí)現(xiàn)

更新時(shí)間：2023年08月09日 16:51:52 作者：弈鳴敲代碼

定時(shí)任務(wù)無(wú)論是個(gè)人開(kāi)發(fā)還是企業(yè)業(yè)務(wù)都是需要的，本文主要介紹了django使用定時(shí)任務(wù)django_apscheduler的實(shí)現(xiàn)，減少請(qǐng)求時(shí)需要用戶等待的時(shí)間，感興趣的可以了解一下

問(wèn)題：當(dāng)數(shù)據(jù)量過(guò)大，請(qǐng)求時(shí)間過(guò)長(zhǎng)的時(shí)候，使用django的緩存可以解決部分問(wèn)題，但是當(dāng)緩存數(shù)據(jù)過(guò)期時(shí)重新請(qǐng)求又是耗時(shí)的時(shí)間，請(qǐng)求時(shí)需要用戶等待。

解決：使用定時(shí)任務(wù)加緩存的方式，在后臺(tái)定時(shí)請(qǐng)求數(shù)據(jù)保存到緩存中，而對(duì)外的接口只負(fù)責(zé)讀取緩存的數(shù)據(jù)即可，這樣就不用再等待數(shù)據(jù)的請(qǐng)求了

django_apscheduler的運(yùn)行和django項(xiàng)目的運(yùn)行是相互獨(dú)立的，即使django項(xiàng)目停止，django_apscheduler也會(huì)繼續(xù)運(yùn)行

一、安裝及配置django_apscheduler

1，使用pip安裝django_apscheduler

pip install -i https://pypi.douban.com/simple django_apscheduler

2，settings.py中配置INSTALLED_APPS

在INSTALLED_APPS中添加django_apscheduler

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'corsheaders',
    'rest_framework',
    'course',
    'django_apscheduler',  # 添加
]

3，生成遷移

python manage.py makemigrationspython manage.py migrate

遷移后生成兩張表

二、安裝及配置django_redis

1，使用pip安裝django_redis

pip install -i https://pypi.douban.com/simple django_redis

2，在settings.py中配置CACHES

# 配置緩存保存
CACHES = {
    'default': {
        'BACKEND': 'django_redis.cache.RedisCache',
        'LOCATION': 'redis://127.0.0.1:6379/1',  # 使用1號(hào)庫(kù)
    }
}

三、代碼

1，定時(shí)請(qǐng)求test_job函數(shù)，請(qǐng)求獲取數(shù)據(jù)保存在redis緩存中

from apscheduler.schedulers.background import BackgroundScheduler
from django_apscheduler.jobstores import DjangoJobStore, register_events, register_job
try:
    scheduler = BackgroundScheduler()
    scheduler.add_jobstore(DjangoJobStore(), "default")
    @register_job(scheduler, "interval", seconds=10)
    def test_job():
        # 定時(shí)每5秒執(zhí)行一次
        count = 0
        courses = Course.objects.values('name')
        for i in range(1000000):
            for course in courses:
                count += len(str(course["name"]))
        cache.set("word_count", count, 10)
        return JsonResponse({"count": count})
    register_events(scheduler)
    # 啟動(dòng)定時(shí)器
    scheduler.start()
except Exception as e:
    print('定時(shí)任務(wù)異常：%s' % str(e))

2，定義接口，獲取保存的數(shù)據(jù)

 class WordCountView(View):
    def get(self, request):
        word_count = cache.get("word_count")
        if word_count:
            return JsonResponse({"count": word_count})

3，配置路由

path('count/', views.WordCountView.as_view())

四、測(cè)試訪問(wèn)

1，啟動(dòng)django

2，查看redis：TTL為過(guò)期時(shí)間，每10秒過(guò)期，因?yàn)槎〞r(shí)任務(wù)也是10秒，所以就是每10秒自動(dòng)刷新一次

3，請(qǐng)求，時(shí)間只耗時(shí)5ms，正常請(qǐng)求在5秒左右

五、常見(jiàn)問(wèn)題

運(yùn)行報(bào)錯(cuò)：Job identifier (course.views.test_job2) conflicts with an existing job

這是因?yàn)楝F(xiàn)有的工作id與數(shù)據(jù)庫(kù)的id相同導(dǎo)致的，手動(dòng)重要命名id即可

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

您可能感興趣的文章: