裝包：

pip install celery  #安裝celery
pip install eventlet #安裝eventlet 協程

程式碼結構圖：

django專案同名資料夾下建立celery.py 檔案，app下建立tasks.py 檔案

配置檔案：settings ---- __init__.py ---- celery.py ---- tasks.py (檔案位置可參考上圖)

--settings.py--
#celery任務配置
UPLOAD_ROOT=os.path.join(BASE_DIR,'static/upload')
CELERY_BROKER_URL
 
='redis://127.0.0.1:6379/'
CELERY_RESULT_BACKEND='redis://127.0.0.1:6379/'
CELERY_RESULT_SERIALIZER='json'

--__init__.py--
#加入絕對引入這個新特性---absolute_import
#將模組中顯式出現的所有字串轉為unicode型別----unicode_literals

from __future__  import absolute_import,unicode_literals
from .celery import app as celery_app


__all__
 
=['celery_app']

----celery.py----


from __future__  import absolute_import,unicode_literals
from celery import Celery
import os

#設定環境變數----對應自己的專案名（Celery_ttoia）
os.environ.setdefault('DJANGO_SETTINGS_MODULE','Celery_ttoia.settings')

#註冊celery
app=Celery('Celery_ttoia')

#繫結配置檔案
app.config_from_object('
 
django.conf:settings',namespace="CELERY")

#自動發現各個app下的tasks.py檔案
app.autodiscover_tasks()

----tasks.py----


from celery.task import task


#自定義非同步任務
@task
def aync_test():
    print('非同步任務')
    return '123'

----views.py----

#呼叫
from rest_framework.views import APIView
from rest_framework.response import Response

from app import tasks  #導包


# Create your views here.
class Dog(APIView):
    def post(self,request):

        res=tasks.aync_test.delay() #呼叫，delay

        print(res.id)
        print(666)
        return Response({'code':200,'msg':'666'})

啟動celery非同步任務：

第一步----要先去開啟redis伺服器，從setting中可以看到我們依賴redis實現非同步任務。

第二部----啟動專案，python manage.py runserver

第三步----啟動celery非同步，celery worker -A Celery_ttoia -l info -P eventlet （重開一個埠啟動）