django使用定时任务django_apscheduler的实现

pip install -i https://pypi.douban.com/simple 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',  # 添加
]

python manage.py makemigrationspython manage.py migrate

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

# 配置缓存保存
CACHES = {
    'default': {
        'BACKEND': 'django_redis.cache.RedisCache',
        'LOCATION': 'redis://127.0.0.1:6379/1',  # 使用1号库
    }
}

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():
        # 定时每5秒执行一次
        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)
    # 启动定时器
    scheduler.start()
except Exception as e:
    print('定时任务异常：%s' % str(e))

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

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