详解配置Django的Celery异步之路踩坑

为什么需要Celery异步处理

在Django的web应用中，有时候我们需要执行一些耗时的任务，例如发送邮件、处理图片、定时任务等等，如果在web请求中直接执行这些任务，会导致web请求阻塞，用户体验极差。因此，我们需要异步执行这些任务，Celery正是为了解决这样的问题而生。

安装和配置Celery

在Django项目中，安装Celery非常简单，只需要使用pip命令即可：

pip install celery

接着，在Django的settings.py中添加以下配置：

CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'

CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_ACCEPT_CONTENT = ['json']

这里的配置表示我们使用Redis作为Celery的任务队列和结果存储的后端，使用json作为消息传递的数据格式。

编写异步任务

在Django项目的app中，我们可以创建一个tasks.py文件用来存放异步任务的代码。例如，以下是一个异步处理图片的任务：

from celery import task
from PIL import Image

@task()
def process_image(image_path):
    im = Image.open(image_path)
    # 这里省略了对图片进行处理的代码
    im.save(image_path)

其中，@task()表示这是一个Celery任务，process_image函数表示该任务执行的方法。

启动Celery worker

在创建完异步任务后，我们需要启动一个Celery worker来接收任务并执行。

在终端中，进入Django项目的根目录，然后执行以下命令：

celery -A your_project_name worker -l info

其中，your_project_name表示你的Django项目名称。Celery会自动搜索项目根目录下的tasks.py文件，并启动一个worker等待接收任务。

触发异步任务

在Django中触发异步任务，可以使用以下方式：

from .tasks import process_image

process_image.delay('path/to/image')

其中，process_image.delay()方法表示异步执行该任务，'path/to/image'表示传递给任务的参数。

踩过的坑

不同Django版本对应的Celery版本不同

在使用Django和Celery的时候，需要留意不同Django版本所需要匹配的Celery版本。例如，Django 2.2.x需要使用Celery 4.x版本，而Django 3.x版本则需要使用Celery 5.x版本。

Redis序列化的错误

在使用Celery过程中，如果启用了Redis作为任务队列和结果存储的后端，那么需要保证传递的数据格式是可序列化的。否则，会出现Redis序列化的错误。因此，在Celery的配置中，需要设置CELERY_TASK_SERIALIZER、CELERY_RESULT_SERIALIZER和CELERY_ACCEPT_CONTENT等参数。

示例1：使用Celery异步发送邮件

为了演示如何使用Celery异步发送邮件，我们可以创建一个异步任务来实现这个功能：

from celery import task
from django.core.mail import EmailMessage

@task()
def send_email(title, content, to_email):
    email = EmailMessage(title, content, to=[to_email])
    email.send()

在定义好异步任务后，我们可以在Django中调用该任务发送邮件：

from .tasks import send_email

send_email.delay('邮件标题', '邮件内容', '[e-mail]')

send_email.delay()表示异步执行该任务发送邮件，避免阻塞web请求。

示例2：使用Celery异步处理数据统计任务

为了演示如何使用Celery异步处理数据统计任务，我们可以创建一个统计访问网站次数的任务：

from celery import task
from datetime import datetime
from .models import Stats

@task()
def count_access():
    today = datetime.now().date()
    stats, created = Stats.objects.get_or_create(date=today)
    stats.visit_count += 1
    stats.save()

在定义好异步任务后，我们可以在Django中调用该任务进行数据统计：

from .tasks import count_access

count_access.delay()

count_access.delay()表示异步执行该任务统计访问次数。由于该任务是定时任务，我们可以使用Celery的beat模块来调度定时执行该任务。