文章目錄

• Celery
• • • 1 定義
    • 2,使用場景
    • 2 使用celery
    • 3 Django + Celery

Celery

1 定義

Celery 是一個簡單、靈活且可靠的，處理大量消息的分布式系統

它是一個專注于實時處理的任務隊列，同時也支援任務排程

中文官網：http://docs.jinkan.org/docs/celery/

線上安裝 sudo pip3 install celery

離線安裝

tar xvfz celery-0.0.0.tar.gz
cd celery-0.0.0
python3 setup.py build
python3 setup.py install
           

2,使用場景

1，任務排程

2，避開阻塞操作

名詞解釋：

broker - 消息傳輸的中間件，生産者一旦有消息發送，将發至broker；【RQ，redis】

backend - 用于存儲消息/任務結果，如果需要跟蹤和查詢任務狀态，則需添加要配置相關

worker - 工作者 - 消費/執行broker中消息/任務的程序

2 使用celery

​ 1, 建立celery應用

#建立 tasks.py 檔案

from celery import Celery
#初始化celery, 指定broker
app = Celery('guoxiaonao', broker='redis://@127.0.0.1/1')

# 建立任務函數
@app.task
def task_test():
    print("task is running....")
    
#在該tasks.py同級目錄下 執行 celery -A tasks worker --loglevel=info
           

​ 2,推送任務

​ 在tasks.py檔案的同級目錄進入 python3 執行 如下代碼

from tasks import task_test
task_test.delay()

           

​ 3,存儲執行結果

​ Celery提供存儲任務執行結果的方案，需借助 redis 或 mysql 或Memcached 等

​ 詳情可見 http://docs.celeryproject.org/en/latest/reference/celery.result.html#module-celery.result

from celery import Celery


app = Celery('demo',
             broker='redis://@127.0.0.1:6379/1'，
             backend='redis://@127.0.0.1:6379/2',
             )

# 建立任務函數
@app.task
def task_test(a, b):
    print("task is running")
    return a + b
           

3 Django + Celery

1，建立項目+應用

#正常指令
django-admin startproject chongci
python manage.py startapp user
           

2，建立celery.py

在settings.py同級目錄下 建立 celery.py檔案

檔案内容如下：

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

# 為celery設定環境變量
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'chongci.settings')



關于os.environ[]和os.getenv()和os.environ.get()三者簡單差別
	這三者都是擷取環境變量鍵對的值的方法。
	os.environ[]表示的是一個dict型的資料，比如os.environ[‘APPDATA’]就表示 'APPDATA’這個key鍵對的值。但如果key不存在，就會報錯，比如	    os.environ[‘1234’]就會報錯。
	而os.getenv()是一個函數，如果不存在就會傳回None。
	基本可以認為os.getenv == os.environ.get

# 建立應用
app = Celery("chongci")
# 配置應用
app.conf.update(
    # 配置broker
    BROKER_URL='redis://:[email protected]:6379/1',
)
# 設定app自動加載任務
app.autodiscover_tasks(settings.INSTALLED_APPS)
           

3, 在應用子產品【user目錄下】建立tasks.py檔案

檔案内容如下：

from chongci.celery import app
import time

@app.task
def task_test():
    print("task begin....")
    time.sleep(10)
    print("task over....")
           

4, 應用視圖編寫；内容如下：

from django.http import HttpResponse
from .tasks import task_test
import datetime

def test_celery(request):
    task_test.delay()
	now = datetime.datetime.now()
    html = "return at %s"%(now.strftime('%H:%M:%S'))
    return HttpResponse(html)
           

5, 分布式路由下添加 test_celery函數對應路由，此過程略

6, 啟動django python3 manage.py runserver

7, 建立 celery worker

​ 在項目路徑下，即chongci 下 執行如下

celery -A chongci worker -l info
           

8，上線時 執行celery指令

在項目路徑下，即chongci 下 執行如下

celery -A chongci worker -l info
           

8，上線時 執行celery指令

nohup指令：如果你正在運作一個程序，而且你覺得在退出帳戶時該程序還不會結束，那麼可以使用nohup指令。該指令可以在你退出帳戶/關閉終端之後繼續運作相應的程序。
nohup celery -A mywiki worker -l info > c.log 2>&1 &