当前位置：知识百科 > 正文

Django中使用Celery的教程详解

更新时间：2026-04-18 12:03 阅读量：138

下面我就来详细讲解一下"Django中使用Celery的教程详解".

Django是一个流行的Web框架,它的开发速度和易用性使它成为了很多Web开发人员的首选.然而,有些任务在请求响应周期内完成可能不太合适,因为它们可能会需要很长时间才能完成,比如发送电子邮件、生成PDF文件、批量导入等.为了避免这些任务阻塞Web应用程序,我们可以使用celery库,通过异步任务来异步处理这些请求.在本篇文章中,我们将详细讲解如何在Django中使用celery,以便更好地管理异步任务.

为了使用Celery,我们需要在Django项目中安装celery库.可以使用pip工具进行安装：

pip install celery

同时,我们还需要安装消息队列（如RabbitMQ或Redis）来存储我们的任务.在本文中,我们以RabbitMQ为例进行讲解.

安装RabbitMQ：

sudo apt-get install rabbitmq-server

在Ubuntu系统中安装RabbitMQ非常简单.安装了RabbitMQ之后,可以使用以下命令启动它：

sudo service rabbitmq-server start

tasks.py 文件

首先,您需要在您的Django应用程序中定义Celery任务. 让我们在我们的myapp应用程序中创建一个 tasks.py文件.该文件应该包含我们的Celery任务定义.

from celery import task

@task()
def send_mail_task(to, subject, body):
# 这里包含你真正的发送邮件逻辑
pass

上面的代码将任务send_mail_task定义为异步任务,该任务接收三个参数.我们将在下文中演示如何调用该任务.

要能够使用异步任务,您需要将该任务添加到应用程序中的INSTALLED_APPS设置中.打开您的Django项目的settings.py文件,让我们将其添加到INSTALLED_APPS中：

INSTALLED_APPS = [
...
'myapp',
...    
]

添加后,我们可以通过在Celery中定义该任务的实例,开始执行该任务. 我们将在下文中介绍如何启动Celery.

celery.py 文件

此时此刻呢,我们需要在我们的Django项目中创建一个名为celery.py的文件（与settings.py在同一级别的目录中）,并编写以下代码：

import os
from celery import Celery

# 设置默认Django设置模块名称
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')

app = Celery('myproject')

# 从所有已安装应用程序的'celery.py'模块中加载任务
app.autodiscover_tasks()

这段代码将自动发现安装您的INSTALLED_APPS设置的应用程序中的所有celery.py模块,并将其导入到Celery中.在此时此刻呢的步骤中,我们将启动Celery并调用我们的任务.

启动Celery以执行我们的任务：

celery -A myproject worker -l info

这将启动Celery并使其在控制台上输出有关任务活动的详细信息. 现在我们可以使用以下代码在调用MongoDB并连接到Celery时执行上面定义的异步任务

from myapp.tasks import send_mail_task

def my_view(request):
to = 'recipient@example.com'
subject = 'This is your subject'
body = 'This is the message body.'
send_mail_task.delay(to=to, subject=subject, body=body)
return HttpResponse('Mail sent')

此时此刻呢,我将提供两个关于Celery的示例来说明如何在Django中使用Celery.

示例1：定时任务

我们可以使用Celery执行定时任务,如清除过期的session.可以通过以下方式实现：

from django.contrib.sessions.models import Session
from celery.task import periodic_task
from datetime import timedelta
from django.utils import timezone

@periodic_task(run_every=timedelta(minutes=5))
def clear_sessions():
Session.objects.filter(expire_date__lt=timezone.now()).delete()

此示例将会启动一个新的任务,即clear_sessions检查是否存储的会话已过期并清除它们.

示例2：长时间任务

有时,一些任务需要很长时间才能完成,比如导入大量数据或发送大量邮件.如果在请求响应周期内处理这些任务,会导致应用程序响应缓慢,所以呢我们需要使用异步任务来异步处理这些请求.这里是一个导入文件的示例：

from celery import task
import csv, os
from myapp.models import MyModel

@task()
def import_data(filename):
# 打开CSV文件并添加到数据库中
f = open(filename, 'r')
reader = csv.reader(f)
for row in reader:
    MyModel.objects.create(name=row[0], age=row[1])

# 删除CSV文件
os.remove(filename)

在本例中,我们定义了import_data函数并将其标记为异步task任务.该函数使用Python内置csv模块打开一个CSV文件,使用循环将每一行添加到Django model 中. 完成后文件将被删除.

我们可以使用以下代码在Django视图中调用此任务：

from myapp.tasks import import_data

def my_view(request):
filename = '/path/to/myfile.csv'
import_data.delay(filename) 
return HttpResponse('File import has begun.')

当该视图被调用时,它将调用import_data方法并使用.delay()方法将任务调度到Celery工作队列上.现在,该任务将在后台运行,不影响Django的响应速度.

以上就是日讯小常识小编为大家整理的Django中使用Celery的教程详解相关主题介绍,如果您觉得小编更新的文章只要能对粉丝们有用，就是我们最大的鼓励和动力,不要忘记讲本站分享给您身边的朋友哦！！

相关文章