django搭建博客之集成celery

為什麼要集成celery

當集成了celery之後，我們可以利用celery的異步任務和定時任務來處理我們django項目邏輯。例如：異步發送郵件和定時清理緩存等，在我的博客我就利用celery等配置完成定時對豆瓣影單書單的爬取，並且可控計劃(循環、定時、時鐘)類型。

django集成celery—demo

如果把celery集成到django，利用celery的異步處理任務就很方便的解決站點需要異步處理的邏輯，需要注意的是最新的celery4.x已經不能再使用django-celery了

• 通過pycharm新建django工程名為celeryproject

• 在celeryproject/celeryproject目錄下新建celery.py文件，內容如下：

  from __future__ import absolute_import, unicode_literals
  from celery import Celery
  import os
  # 配置默認的django settings模塊配置給celery
  os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'celeryproject.settings')
  app = Celery('celeryproject')
  # 命名空間
  namespace='CELERY'定義所有與celery相關的配置的鍵名要以'CELERY_'為前綴。
  app.config_from_object('django.conf:settings', namespace='CELERY')
  # 如果你把所有的task都定義在單獨的tasks.py模塊中，加上這句話celery會自動發現這些模塊中的task，
  app.autodiscover_tasks()
  

• 添加celeryproject/celeryproject目錄的__init__.py文件內容如下：

  from __future__ import absolute_import, unicode_literals
  from .cerely import app as celery_app
  __all__ = ['celery_app']
  

• 在celeryproject/celeryproject目錄下的settings.py文件添加以下內容：

  CELERY_BROKER_URL = 'redis://127.0.0.1:6379/0'
  CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/1'
  

• 執行python manage.py startapp myapp新建一個app

• 在celeryproject/myapp目錄下新建tasks.py文件內容如下：

  from celery import shared_task
  import time
  @shared_task
  def send_email():
  print("模擬發送郵件")
  import time
  #添加延時是為了模擬發送郵件的耗時
  time.sleep(10)
  return "郵件發送成功"
  

• 在myapp目錄下的view.py裡面添加內容如下：

  from django.http import HttpResponse
  from myapp.tasks import send_email
  def index(request):
  result=send_email.delay()
  return HttpResponse("ok!")
  

• 在celeryproject/celeryproject目錄下urls.py文件配置路由如下

  from django.contrib import admin
  from django.urls import path
  from myapp import views
  urlpatterns = [
  path('admin/', admin.site.urls),
  path('index/',views.index),
  ]
  

• 打開terminal窗口，在根目錄下執行命令celery worker -A celeryproject -l info -P eventlet，出現界面如下，代表celery啟動正常
  

• 運行django工程，浏覽器訪問http://127.0.0.1:8000/index/
  
  由於在模擬發送郵件的代碼添加了時間等待，可以很明顯的感覺到浏覽器內容ok顯示出來之後，回到terminal窗口，過了一會兒才提示郵件發送成功，說明任務的確是異步處理的

django集成django_celery_reuslts

• 安裝

  pip install celery django-celery-results
  

• 添加應用到django

   INSTALLED_APPS = [
  ...
  'django_celery_results.apps.CeleryResultConfig',
  ]
  

• 在項目的settings.py更改存儲結果後台為數據庫存儲

  CELERY_RESULT_BACKEND = 'django-db' #存儲任務結果
  

• 執行數據庫遷移命令

  python manage.py makemigrations
  python manage.py migrate
  

當遷移完畢，django_celery_reuslts會生成django_celery_results_taskresult表，如果有任務執行完畢，django_celery_results會將數據結果存儲到該表，大致表內容如下：

django集成django_celery_beat

• 安裝

  pip install django-celery-beat
  

• 添加應用到django

  INSTALLED_APPS = [
  ...
  'django_celery_results.apps.CeleryResultConfig',
  'django_celery_beat.apps.BeatConfig',
  ]
  

• 修改項目的settings.py裡面的內容為如下

  CELERY_TIMEZONE = TIME_ZONE #保持時區跟當前項目時區一致
  # CELERY_ENABLE_UTC=False
  DJANGO_CELERY_BEAT_TZ_AWARE = False #解決時區問題
  CELERY_BROKER_URL = 'redis://127.0.0.1:6379/0'
  # CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/0' #redis存儲任務結果
  CELERY_RESULT_BACKEND = 'django-db' #存儲任務結果
  CELERY_TASK_SERIALIZER = 'pickle'
  CELERY_RESULT_SERIALIZER = 'pickle'
  CELERY_ACCEPT_CONTENT = ['pickle', 'json']
  CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'
  

• 執行數據庫遷移命令

  python manage.py makemigrations
  python manage.py migrate
  

當遷移完畢，django_celery_reuslts會生成五張表（前面3張表比較重要）

表名描述django_celery_beat_clockedschedule周期性任務表django_celery_beat_intervalschedule間隔時間計劃表（可循環）django_celery_beat_crontabschedule定時計劃表django_celery_beat_clockedschedule時鐘計劃表(單次執行)django_celery_beat_solarschedule經緯度表計劃表django_celery_beat_periodictask計劃何時更改記錄表

而且由於我們已經集成django_celery_reuslts，如果我們通過django_celery_beat後台配置任務時，當任務執行完畢，django_celery_results也會將數據結果存儲到表裡面

測試django-celery-beat任務

1. 在django的應用目錄下新建tasks.py文件

   from __future__ import absolute_import
   from celery import shared_task
   @shared_task
   def test1():
   print("測試間歇性任務")
   return "間歇性任務ok"
   @shared_task
   def test2():
   print("測試定時任務")
   return "定時任務ok"
   @shared_task
   def test3():
   print("測試時鐘任務")
   return "時鐘任務ok"
   

2. 啟動django訪問後台
   我這裡用的是xamdin，我這裡將django-celery-beat的模型注冊到下xadmin了
   

3. 本地開啟redis服務（一定要記得開啟）

4. 執行命令 celery worker -A tasks -l info -P eventlet
   可以看到任務已經被celery識別到
   

• 在後台新建intervalschedule、crontabschedule、clockedschedule三個任務
  重點
  1.每個任務任務只能選取一種schedule類型，否則會報錯的
  2.taskname要填寫完整的任務路徑
  3.由於clockedschedule存在時區差8小時問題，故設置任務執行的時間必須要減去8小時才會准時執行。
  
• 任務創建後如下圖
  
• 執行命令celery beat -A blog -l info
  可以看到間歇任務已經開始發送了，切回到worker的terminal界面可以看到如下界面
• 間歇性任務執行結果
  
• 開始發送定時任務
  
• 定時任務執行結果
  
• 開始發送時鐘任務
  
• 時鐘任務執行結果（時鐘任務只會執行一次，執行完畢計劃表就會過期）
  

關於celery任務模塊識別探討

下面讓我們來測試下，我在django的任意的應用目錄下新建tasks.py文件，內容如下

 from celery import shared_task
@shared_task
def test3():
print("測試django應用下的tasks.py裡面任務是否可以識別出來")

然後我在應用目錄下的tests.py下新建任務如下：

from celery import shared_task
@shared_task
def test2():
print("測試django應用下tests.py裡面的任務是否可以識別")

用執行下celery命令看下任務列表

celery worker -A blog -l info -P eventlet

結果如下：

[tasks]
. apps.util.tasks.test3

從上面可以知道celery會自動發現django應用目錄下tasks.py裡面的任務，不是tasks.py裡面的任務則無法識別
不過我又發現了一個很奇怪的問題，比方說在任意應用的views.py裡面這樣

from apps.util.tests import test2

再執行下celery命令看下任務列表

celery worker -A blog -l info -P eventlet

結果如下：

[tasks]
. apps.util.tasks.test3
. apps.util.tests.test2

發現celery又能識別出任務了。我猜測這樣的操作應該是變相的完成了celery的CELERY_IMPORTS配置，我們取消上面剛剛的導入，在項目的settings.py裡面增加配置如下：

CELERY_IMPORTS=["apps.util.test"]

再執行下celery命令看下任務列表

celery worker -A blog -l info -P eventlet

結果如下：

[tasks]
. apps.util.tasks.test3
. apps.util.tests.test2

也就是說任務不一定要放在tasks.py文件內,並且任務模塊不一定要在配置文件裡面注明，項目裡面完成了該模塊導入都行。