1.創建django項目delery_demo, 創建應用demo:
django-admin startproject delery_demo 創建項目
python manage.py startapp demo 創建app
2.項目的結構
3.在celery_demo模塊中創建celery.py模塊, 文件目錄為:
4.celery.py模塊內容為:
from celery import Celery from django.conf import settings import os # 為celery設置環境變量 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'celery_demo.settings') # 創建應用 app = Celery("demo") # 配置應用 app.conf.update( # 配置broker, 這里我們用redis作為broker BROKER_URL='redis://:332572@127.0.0.1:6379/1', ) # 設置app自動加載任務 # 從已經安裝的app中查找任務 app.autodiscover_tasks(settings.INSTALLED_APPS)
5.在應用demo引用創建tasks.py模塊, 文件目錄為:
6.我們在文件內創建一個任務函數my_task:
from celery_demo.celery import app import time # 加上app對象的task裝飾器 # 此函數為任務函數 @app.task def my_task(): print("任務開始執行....") time.sleep(5) print("任務執行結束....")
7.在views.py模塊中創建視圖index:
from django.shortcuts import render from django.http import HttpResponse from .tasks import my_task def index(request): # 將my_task任務加入到celery隊列中 # 如果my_task函數有參數，可通過delay()傳遞 # 例如 my_task(a, b), my_task.delay(10, 20) my_task.delay() return HttpResponse("<h1>服務器返回響應內容!</h1>")
8.在celey_demo/settings.py配置視圖路由:
from django.conf.urls import url from django.contrib import admin from demo.views import index urlpatterns = [ url(r'^admin/', admin.site.urls), url(r'^$', index), ]
9.創建worker等待處理celery隊列中任務, 在終端執行命令:
celery -A celery_demo worker -l info
結果
10.啟動django測試服務器：
python manage.py runserver
11.此處需要用到額外包django_celery_results, 先安裝包:
pip install django-celery-results
12.在celery_demo/settings.py中安裝此應用:
INSTALLED_APPS = [ 'django.contrib.admin', 'django.contrib.auth', 'django.contrib.contenttypes', 'django.contrib.sessions', 'django.contrib.messages', 'django.contrib.staticfiles', 'demo', 'django_celery_results', # 注意此處應用名為下划線 ]
13.回到celery_demo/celery.py模塊中，增加配置信息如下:
from celery import Celery from django.conf import settings import os # 為celery設置環境變量 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'celery_demo.settings') # 創建應用 app = Celery("demo") # 配置應用 app.conf.update( # 配置broker, 這里我們用redis作為broker BROKER_URL='redis://:332572@127.0.0.1:6379/1', # 使用項目數據庫存儲任務執行結果 CELERY_RESULT_BACKEND='django-db', ) # 設置app自動加載任務 # 從已經安裝的app中查找任務 app.autodiscover_tasks(settings.INSTALLED_APPS)
14.創建django_celery_results應用所需數據庫表, 執行遷移文件：
python manage.py migrate django_celery_results
15.如果我們想某日某時執行某個任務，或者每隔一段時間執行某個任務，也可以使用celery來完成. 使用定時任務，需要安裝額外包:
pip install django_celery_beat
16.首先在settings.py中安裝此應用:
INSTALLED_APPS = [ 'django.contrib.admin', 'django.contrib.auth', 'django.contrib.contenttypes', 'django.contrib.sessions', 'django.contrib.messages', 'django.contrib.staticfiles', 'demo', 'django_celery_results', 'django_celery_beat', # 安裝應用 ]
17.在celery_demo/celery.py模塊中增加定時任務配置:
from celery import Celery from django.conf import settings import os # 為celery設置環境變量 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'celery_demo.settings') # 創建應用 app = Celery("demo") # 配置應用 app.conf.update( # 配置broker, 這里我們用redis作為broker BROKER_URL='redis://:332572@127.0.0.1:6379/1', # 使用項目數據庫存儲任務執行結果 CELERY_RESULT_BACKEND='django-db', # 配置定時器模塊，定時器信息存儲在數據庫中 CELERYBEAT_SCHEDULER='django_celery_beat.schedulers.DatabaseScheduler', ) # 設置app自動加載任務 # 從已經安裝的app中查找任務 app.autodiscover_tasks(settings.INSTALLED_APPS)
18.由於定時器信息存儲在數據庫中，我們需要先生成對應表, 對diango_celery_beat執行遷移操作，創建對應表:
python manage.py migrate django_celery_beat
19.我們可登錄網站后台Admin去創建對應任務, 首先我們先在tasks.py模塊中增加新的任務，用於定時去執行(5秒執行一次)
from celery_demo.celery import app import time # 用於定時執行的任務 @app.task def interval_task(): print("我每隔5秒鍾時間執行一次....")
20.首先創建后台管理員帳號:
python manage.py createsuperuser
21.登錄管理后台Admin:
22.其中Crontabs用於定時某個具體時間執行某個任務的時間，Intervals用於每隔多久執行任務的事件，具體任務的執行在Periodic tasks表中創建。
我們要創建每隔5秒執行某個任務，所以在Intervals表名后面點擊Add按鈕:
23.然后在Periodic tasks表名后面，點擊Add按鈕，添加任務:
24.啟動定時任務：
celery -A celery_demo worker -l info --beat
人生忽如寄，山水有相逢