IvanKao的部落格

Python Django 實作：Django + Celery + Redis 異步執行任務

一、概要：

為了增進使用者的體驗，常常會需要使用異步的方式來處理任務

可以使用Celery在後台異步運行一段程式，讓你的應用程式可以繼續響應客戶端的請求

例如發送信件不使用異步更新，就可能要稍等幾秒鐘將信件寄出，使用者才能繼續操作

如果使用Celery異步更新的話，發送信件則會在後台異步運行，使用者不需等待即可繼續操作

而這次，我則是要使用Celery在異步運行爬蟲程式，幫助我在後台定期爬取資料

二、準備：

1.安裝 Celery

(1)在windows安裝 Celery

在Windows 安裝 Celery 不支援4.0以上的版本

所以這裡安裝3.18版本

打開Django專案，並使用pip安裝Celery

pip install celery==3.1.18

(2)在 Linux 安裝 Celery

在 Linux 安裝 Celery

pip3 install celery

2.安裝 Redis

(1)在windows安裝 Redis

在 此github 下載windows適用的Redis 的msi檔案，並安裝

安裝後可以開啟工作管理員→服務，查看是否有正確運行

將下列代碼添加至Django專案的<settings.py>

#redis
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 = 'Asia/Taipei'

測試Celery是否正確安裝，請在專案環境下輸入以下指令：

celery -A 專案目錄 worker -l info

我的專案目錄名稱為savemoney，以下是我的運行結果

用ctrl+c退出，並輸入以下代碼確認調度任務程序是否已經準備完成：

celery -A 專案目錄 beat -l info

我的專案目錄名稱為savemoney，以下是我的運行結果

(2)在Linux安裝 Redis

1.安裝

$ sudo apt-get update 
$ sudo apt-get install redis-server

啟動 redis-server ：$ redis-server

查看 redis ：$ redis-cli

2.使用 django-redis

在虛擬環境中輸入指令進行安裝： pip3 install django-redis

將下列代碼添加至Django專案的<settings.py>

#redis
CELERY_BROKER_URL = 'redis://localhost:6379'
#: Only add pickle to this list if your broker is secured
CELERY_ACCEPT_CONTENT = ['json']
CELERY_RESULT_BACKEND = 'redis://localhost:6379'
CELERY_TASK_SERIALIZER = 'json'
CELERY_ENABLE_UTC = True
CELERY_TIMEZONE = 'Asia/Taipei'

三、使用Celery

1.在Windos配置Celery 3.1.8

在專案目錄創建celery.py，並添加以下內容：

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' program. 
os.environ.setdefault('DJANGO_SETTINGS_MODULE', '<mysite>.settings') 
app = Celery('<mysite>') 
 
# 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))

在專案根目錄的專案資料夾的__init__.py添加以下代碼：

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

在settings.py添加：

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 = 'Asia/Taipei'

2.在Linux配置Celery 4.2.1

在專案目錄創建celery.py，並添加以下內容：

# http://docs.celeryproject.org/en/latest/django/first-steps-with-django.html
from __future__ import absolute_import, unicode_literals
import os
from celery import Celery
os.environ.setdefault('DJANGO_SETTINGS_MODULE', '<mysite>.settings')

app = Celery('<mysite>')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

在專案根目錄的專案資料夾的__init__.py添加以下代碼：

from __future__ import absolute_import, unicode_literals

# This will make sure the app is always imported when
# Django starts so that shared_task will use this app.
from .celery import app as celery_app

__all__ = ['celery_app']

在settings.py添加：

# Celery settings
CELERY_BROKER_URL = 'redis://localhost:6379'
#: Only add pickle to this list if your broker is secured
CELERY_ACCEPT_CONTENT = ['json']
CELERY_RESULT_BACKEND = 'redis://localhost:6379'
CELERY_TASK_SERIALIZER = 'json'
CELERY_ENABLE_UTC = True
CELERY_TIMEZONE = 'Asia/Taipei'

3.Celery任務

接下來將一個基本函數做一個簡單的Celery任務

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

導入Celery相關套件與使用裝飾器

from celery.decorators import task 

@task(name="sum_two_numbers") 
def add(x, y): 
    return x + y

接著就透過以下方式使用Celery異步執行：

add.delay(7, 8)

4.週期任務

在Django要執行Celery的目錄創建一個<task.py>文件

並添加以下套件，最後則是導入要執行的自訂函數與獲取應用名稱：

from celery.decorators import task 
from celery.task.schedules import crontab
from celery.decorators import periodic_task
from celery.utils.log import get_task_logger

from yourapp.crawler import *
logger = get_task_logger(__name__)  

接著添加裝飾器，這邊是設置每10分鐘運行一次，設置詳情可以參照此處

還有運行的任務名稱：

@periodic_task(  
    run_every=(crontab(minute='*/10')),
    name="some_task",  
    ignore_result=True  
)

最後則是添加任務：

def some_task():
    
    crawler() #此為調用的自訂函數名稱
    logger.info("crawled_momo")

5.本地運行

在Django和Redis運行的環境下

在虛擬環境的專案目錄運行Celery指令，每個指令都需要新開一個命令窗口獨立運行

我的專案目錄名稱為savemoney，以下為代碼：

celery -A savemoney worker -l info 
celery -A savemoney beat -l info

下圖兩圖我自己的爬蟲task運行後的情形：

有兩個task，一個為momo_task，一個為pchome_task：