Django Celery Redis 異步執行任務demo實例

1、windows中安裝redis

 安裝過程見 《在windows x64上部署使用Redis》

2、環境準備

requirements.txt

Django==1.10.5
celery==3.1.23
redis==2.10.5

 注意，celery 4.x 以上不支持windows

pip install -r requirements.txt

3、建立Django項目celery_proj,建立APP:celery_demo

>>django-admin startproject celery_proj
>>cd celery_proj
>>django-admin startapp celery_demo

4、添加celery相關配置信息

1.在celery_proj/celery_proj目錄下，添加以下 celery.py 文件

#!/usr/bin/env python 
# -*- coding: utf-8 -*-

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

# set the default Django settings module for the 'celery_proj' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'celery_proj.settings')
app = Celery('celery_proj')

# Using a string here means the worker will not have to
# pickle the object when using Windows.
app.config_from_object('django.conf:settings')
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)


@app.task(bind=True)
def debug_task(self):
    print('Request: {0!r}'.format(self.request))

2.在celery_proj/celery_proj/__init__.py 中添加：

from .celery import app as celery_app

3.在項目配置文件中，如celery_proj/celery_proj/settings.py 中添加：

# CELERY STUFF
BROKER_URL = 'redis://localhost:6379'
CELERY_RESULT_BACKEND = 'redis://localhost:6379'
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Africa/Nairobi'

5、建立異步運行任務tasks

1.在APP目錄下，如celery_proj/celery_demo目錄下添加以下 tasks.py

#!/usr/bin/env python 
# -*- coding: utf-8 -*-

import time
from celery import task


@task()
def add(x, y):
    return x + y


@task
def run_test_suit(ts_id):
    print "++++++++++++++++++++++++++++++++++++"
    print('jobs[ts_id=%s] running....' % ts_id)
    time.sleep(10.0)
    print('jobs[ts_id=%s] done' % ts_id)
    result = True
    return result

2.在views視圖中，添加相關視圖，如celery_proj/celery_demo/views.py:

from django.http import HttpResponse
from .tasks import run_test_suit


def tasks(request):
    print('before run_test_suit')
    result = run_test_suit.delay('110')
    print('after run_test_suit')
    return HttpResponse("job is runing background~")

3.添加url映射，如修改 celery_proj/celery_proj/urls.py：

from django.conf.urls import url
from celery_demo import views

urlpatterns = [
    url(r'^', views.tasks, name='task'),
]

4.啓動django server

>>python manage.py runserver

5.啓動celery worker

>>celery -A celery_proj worker -l info

當顯示以下，證實worker已啓動成功：

E:\workspace\celery_proj>celery -A celery_proj worker -l info

 -------------- celery@WINDOWS-QKKYZQB v3.1.23 (Cipater)
---- **** -----
--- * ***  * -- Windows-7-6.1.7601-SP1
-- * - **** ---
- ** ---------- [config]
- ** ---------- .> app:         celery_proj:0x33fc190
- ** ---------- .> transport:   redis://localhost:6379//
- ** ---------- .> results:     redis://localhost:6379/
- *** --- * --- .> concurrency: 4 (prefork)
-- ******* ----
--- ***** ----- [queues]
 -------------- .> celery           exchange=celery(direct) key=celery


[tasks]
  . celery_demo.tasks.add
  . celery_demo.tasks.run_test_suit
  . celery_proj.celery.debug_task

[2017-03-16 18:32:47,885: INFO/MainProcess] Connected to redis://localhost:6379//

注意：每當tasks.py 修改時，都要從新啓動worker

6.訪問瀏覽器：http://localhost:8000/

django控制檯立刻輸出相關打印語句而且返回 HttpResponse 響應

可是實際上 run_test_suit 任務未完成，任務在後臺運行 模擬sleep 10s後才完成，運行結果可經過celery worker 控制檯輸出中顯示。

 

---

***微信掃一掃，關注「python測試開發圈」，瞭解更多測試教程！***