Django中使用Celery（通用方案、官方方案）

通用方案

场景

现在定义了一张图片表，要求每隔一段时间就将表内容更新进缓存库中

前置准备

• 安装模块

pip install Django4.2.13
pip install celery
pip install cache
pip install eventlet
pip install redis

• 定义图片表

from django.db import models

class Display(models.Model):
    title = models.CharField(max_length=16, unique=True, verbose_name='名称')
    image = models.ImageField(upload_to='banner', verbose_name='图片')
    link = models.CharField(max_length=64, verbose_name='跳转链接')
    info = models.TextField(verbose_name='详情')

• 路径结构

完整代码

# tasks.py
from celerys import app
from datetime import timedelta

app.conf.beat_schedule = {
    'update_cache_every_30_seconds': {
    'task': 'celerys.update_cache',
    'schedule': timedelta(seconds=30),
    },
}

# celerys.py
import os
import sys

# 执行文件必须与Django配置的环境变量同级(manage.py)
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'Luffy_API.settings.dev')
import django
django.setup()

from app.models import Display
from app.serializer import DisplaySerializer
from celery import Celery
from celery.schedules import crontab
from datetime import timedelta
from django.core.cache import cache

broker = 'redis://127.0.0.1:6379/1'
backend = 'redis://127.0.0.1:6379/2'
app = Celery('celery',broker=broker,backend=backend)

@app.task
def update_cache():
    queryset = Display.objects.all()
    serializer = DisplaySerializer(instance=queryset, many=True)
    cache_data = queryset.values()
    cache.set('display', cache_data)
    return '缓存已更新'

• 执行命令

celery -A celerys worker -l info -P eventlet
celery -A tasks beat -l info

Celery官方方案

【1】注册celery配置

在settings中添加Celery的配置文件

# 用redis创建队列库1
CELERY_BROKER_URL='redis://127.0.0.1:6379/1'
# 用redis创建结果存储库2
CELERY_RESULT_BACKEND='redis://127.0.0.1:6379/2'
CELERY_ACCEPT_CONTENT=['json']
CELERY_TASK_SERIALIZER='json'
# 结果序列化方案
CELERY_RESULT_SERIALIZER='json'
# 任务结果过期时间，秒
CELERY_TASK_RESULT_EXPIRES=60 * 60 * 24
# 时区配置
CELERY_TIMEZONE='Asia/Shanghai'

【2】创建celery文件

新建tasks文件和celerys文件

• 目录结构

【3】init注册

在根目录下的__init__.py文件中注册celerys文件，如果不加则会自动识别根目录中的celery文件而不是我们创建的celerys

from .celerys import app as celery_app

__all__=('celery_app',)

【4】添加任务

tasks.py文件中存储要被执行的异步文件

from celerys import app

@add.task()
def add(a, b):
    return a + b

celerys.py存放celery的配置文件

import os
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'Django_API.settings.dev')
import django
from celery import Celery, shared_task
from django.conf import settings

# 启动worker的命名
app = Celery('Django_API')

# 自动检索前缀为CELERY的配置
app.config_from_object('django.conf:settings', namespace='CELERY')

# 自动搜索每个app中的tasks.py文件
app.autodiscover_tasks()

【5】启动worker异步任务

启动worker

celery -A Django_API worker -l info -P eventlet

【6】在view中调用

from rest_framework.views import APIView
from rest_framework.response import Response
from .tasks import add

class TestView(APIView):
    def get(self, request):
        res = add.delay(500, 2)
        return Response(f'res:{res}')