Django中celery的使用

时间：2021-01-05 15:45:44 阅读：27 评论：0 收藏：0 [点我收藏+]

这里主要展示一个最简单的django中的celery任务，为了让大家都可以用上celery。话不多说，首先给大家看一下我的目录

技术分享图片这里的TestCelery是我的项目名称，CeleryTask是app名称。

这个项目安装的环境：(感觉有几个没用到，反正先装上)

(1) Python == 3.5.2 (2) Django == 2.1.15 (3) Celery == 3.1.26.post2 (4) Django-Celery == 3.2.2 (5) Django-redis == 4.10.0 (6) redis == 2.10.6

我这里用到的中间人Broker是redis，所以就安装了redis相关的包。接下去就是操作步骤。

----------------------------------------------------------------------------------------------------------------------------------

Step1：修改settings.py，在最后加上这两句。(看了网上很多教程，要加INSTALL APP, 以及一堆配置。但后来我一个个删掉，发现最后剩这两个就可以运行最简单的Celery了)

BROKER_URL = ‘redis://127.0.0.1:6379/14‘  # 任务容器地址，redis数据库地址
CELERY_RESULT_BACKEND = ‘redis://127.0.0.1:6379/15‘  # 任务结束的地址

Step2：在TestCelery下新建一个celery.py

import os
from celery import Celery
from django.conf import settings

# 设置celery的环境变量和django-celery的工作目录
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "TestCelery.settings")
# 实例化celery应用，传入服务器名称
app = Celery("TestCelery")
# 加载celery配置
app.config_from_object("django.conf:settings")

# 如果在项目中，创建了task.py,那么celery就会沿着app去查找task.py来生成任务
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

Step3：在TestCelery下的__init__.py

from .celery import app as celery_app
__all__ = [‘celery_app‘]

Step4：在CeleryTask下新建tasks.py

from __future__ import absolute_import, unicode_literals

import time

from celery import shared_task

@shared_task
def adds():
    for i in range(0, 10):
        print(i)
        time.sleep(1)
    return ‘finish‘

Step5：在TestCelery下的views.py

from django.http import JsonResponse, HttpResponse
from CeleryTask import tasks
from celery.result import AsyncResult


# Create your views here.

def index(request):
    """ 进入这个url的时候就触发异步任务，并在session中记录task_id """
    res = tasks.adds.delay()
    request.session[‘task_id‘] = res.task_id
    return JsonResponse({‘status‘: ‘successful‘, ‘task_id‘: res.task_id})


def dasd(request):
    """ 进入url就会去获取session中的task_id，并检测任务. 若任务还在进行就显示页面还在加载，若进行完成就显示hahaha """
    task_id = request.session.get(‘task_id‘)
    if task_id and AsyncResult(task_id).state == ‘PENDING‘:  # 加载时的状态为PENDING
        return HttpResponse(‘页面正在加载...‘)
    return HttpResponse(‘hahaha‘)

Step6：在TestCelery下的urls.py

from django.conf.urls import url
from django.contrib import admin
from TestCelery import views

urlpatterns = [
    url(r‘^admin/‘, admin.site.urls),
    url(r‘^index/‘, views.index),
    url(r‘^dasd/‘, views.dasd),
]

代码部分结束

---------------------------------------------------------------------------------------------------------------------------------------------------------

开启程序部分：

Step1：开启本地的redis

Step2：开启django项目

Step3：在Terminal中输入 celery worker -A TestCelery -l INFO

终端里显示的界面如下：

技术分享图片