Celery異步處理

時間 2019-11-10 標籤 celery 異步處理

1.Celery概述

1.1問題拋出

咱們在作網站後端程序開發時，會碰到這樣的需求：用戶須要在咱們的網站填寫註冊信息，咱們發給用戶一封註冊激活郵件到用戶郵箱，若是因爲各類緣由，這封郵件發送所需時間較長，那麼客戶端將會等待好久，形成很差的用戶體驗.html

那麼怎麼解決這樣的問題呢?python

咱們將耗時任務放到後臺異步執行。不會影響用戶其餘操做。除了註冊功能，例如上傳，圖形處理等等耗時的任務，均可以按照這種思路來解決。如何實現異步執行任務呢？web

咱們可以使用celery. celery除了剛纔所涉及到的異步執行任務以外，還能夠實現定時處理某些任務。redis

1.2celery介紹

Celery是一個功能完備即插即用的任務隊列。它使得咱們不須要考慮複雜的問題，使用很是簡單。celery看起來彷佛很龐大，本章節咱們先對其進行簡單的瞭解，而後再去學習其餘一些高級特性。 
celery適用異步處理問題，當發送郵件、或者文件上傳, 圖像處理等等一些比較耗時的操做，咱們可將其異步執行，這樣用戶不須要等待好久，提升用戶體驗。 celery的特色是：

簡單，易於使用和維護，有豐富的文檔。
高效，單個celery進程每分鐘能夠處理數百萬個任務。
靈活，celery中幾乎每一個部分均可以自定義擴展。
celery很是易於集成到一些web開發框架中.

二、快速入門（一）

2.1任務隊列

任務隊列是一種跨線程、跨機器工做的一種機制.sql

任務隊列中包含稱做任務的工做單元。有專門的工做進程持續不斷的監視任務隊列，並從中得到新的任務並處理.數據庫

celery經過消息進行通訊，一般使用一個叫Broker(中間人)來協client(任務的發出者)和worker(任務的處理者). clients發出消息到隊列中，broker將隊列中的信息派發給worker來處理。django

一個celery系統能夠包含不少的worker和broker，可加強橫向擴展性和高可用性能。後端

2.2Celery安裝

咱們可使用python的包管理器pip來安裝:

pip install -U Celery
也可從官方直接下載安裝包:https://pypi.python.org/pypi/celery/

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

2.3Broker(中間人)

Celery須要一種解決消息的發送和接受的方式，咱們把這種用來存儲消息的的中間裝置叫作message broker, 也可叫作消息中間人。做爲中間人，咱們有幾種方案可選擇：瀏覽器

RabbitMQ

RabbitMQ是一個功能完備，穩定的而且易於安裝的broker. 它是生產環境中最優的選擇。使用RabbitMQ的細節參照如下連接： http://docs.celeryproject.org/en/latest/getting-started/brokers/rabbitmq.html#broker-rabbitmq服務器

若是咱們使用的是Ubuntu或者Debian發行版的Linux，能夠直接經過下面的命令安裝RabbitMQ: sudo apt-get install rabbitmq-server 安裝完畢以後，RabbitMQ-server服務器就已經在後臺運行。若是您用的並非Ubuntu或Debian, 能夠在如下網址： http://www.rabbitmq.com/download.html 去查找本身所須要的版本軟件。

Redis

Redis也是一款功能完備的broker可選項，可是其更可能因意外中斷或者電源故障致使數據丟失的狀況。關因而有那個Redis做爲Broker，可訪下面網址： http://docs.celeryproject.org/en/latest/getting-started/brokers/redis.html#broker-redis

2.4應用

使用celery第一件要作的最爲重要的事情是須要先建立一個Celery實例，咱們通常叫作celery應用，或者更簡單直接叫作一個app。app應用是咱們使用celery全部功能的入口，好比建立任務，管理任務等，在使用celery的時候，app必須可以被其餘的模塊導入。

建立應用

咱們首先建立tasks.py模塊, 其內容爲:

from celery import Celery # 咱們這裏案例使用redis做爲broker app = Celery('demo', broker='redis://:332572@127.0.0.1/1') # 建立任務函數 @app.task def my_task(): print("任務函數正在執行....")

Celery第一個參數是給其設定一個名字，第二參數咱們設定一箇中間人broker, 在這裏咱們使用Redis做爲中間人。my_task函數是咱們編寫的一個任務函數，經過加上裝飾器app.task, 將其註冊到broker的隊列中。

如今咱們在建立一個worker，等待處理隊列中的任務.打開終端，cd到tasks.py同級目錄中，執行命令:

celery -A tasks worker --loglevel=info

顯示效果以下:

調用任務

任務加入到broker隊列中，以便剛纔咱們建立的celery workder服務器可以從隊列中取出任務並執行。如何將任務函數加入到隊列中，可以使用delay()。

進入python終端, 執行以下代碼:

from tasks import my_task
my_task.delay()

執行效果以下:

咱們經過worker的控制檯，能夠看到咱們的任務被worker處理。調用一個任務函數，將會返回一個AsyncResult對象，這個對象能夠用來檢查任務的狀態或者得到任務的返回值。

存儲結果

若是咱們想跟蹤任務的狀態，Celery須要將結果保存到某個地方。有幾種保存的方案可選:SQLAlchemy、Django ORM、Memcached、 Redis、RPC (RabbitMQ/AMQP)。

例子咱們仍然使用Redis做爲存儲結果的方案，任務結果存儲配置咱們經過Celery的backend參數來設定。咱們將tasks模塊修改以下:

from celery import Celery # 咱們這裏案例使用redis做爲broker app = Celery('demo', backend='redis://:332572@127.0.0.1:6379/2', broker='redis://:332572@127.0.0.1:6379/1') # 建立任務函數 @app.task def my_task(a, b): print("任務函數正在執行....") return a + b

咱們給Celery增長了backend參數，指定redis做爲結果存儲,並將任務函數修改成兩個參數，而且有返回值。

更多關於result對象信息，請參閱下列網址:http://docs.celeryproject.org/en/latest/reference/celery.result.html#module-celery.result

2.5配置

Celery使用簡單，配置也很是簡單。Celery有不少配置選項可以使得celery可以符合咱們的須要，可是默認的幾項配置已經足夠應付大多數應用場景了。

配置信息能夠直接在app中設置，或者經過專有的配置模塊來配置。

直接經過app來配置

from celery import Celery app = Celery('demo') # 增長配置 app.conf.update( result_backend='redis://:332572@127.0.0.1:6379/2', broker_url='redis://:332572@127.0.0.1:6379/1', )

專有配置文件

對於比較大的項目，咱們建議配置信息做爲一個單獨的模塊。咱們能夠經過調用app的函數來告訴Celery使用咱們的配置模塊。

配置模塊的名字咱們取名爲celeryconfig, 這個名字不是固定的，咱們能夠任意取名，建議這麼作。咱們必須保證配置模塊可以被導入。配置模塊的名字咱們取名爲celeryconfig, 這個名字不是固定的，咱們能夠任意取名，建議這麼作。咱們必須保證配置模塊可以被導入。

下面咱們在tasks.py模塊同級目錄下建立配置模塊celeryconfig.py:

result_backend = 'redis://:332572@127.0.0.1:6379/2' broker_url = 'redis://:332572@127.0.0.1:6379/1'

tasks.py文件修改成:

from celery import Celery import celeryconfig # 咱們這裏案例使用redis做爲broker app = Celery('demo') # 從單獨的配置模塊中加載配置 app.config_from_object('celeryconfig')

三、快速入門（二）

3.1項目中使用celery

  個人項目目錄:

TestCelery/ ├── proj │ ├── celeryconfig.py │ ├── celery.py │ ├── init.py │ └── tasks.py └── test.py

  celery.py內容以下:

from celery import Celery

# 建立celery實例
app = Celery('demo')
app.config_from_object('proj.celeryconfig')

# 自動搜索任務
app.autodiscover_tasks(['proj'])
  celeryconfig.p模塊內容以下：

from kombu import Exchange, Queue
BROKER_URL = 'redis://:332572@127.0.0.1:6379/1'
CELERY_RESULT_BACKEND = 'redis://:332572@127.0.0.1:6379/2'
  tasks.py模塊內容以下:

from proj.celery import app as celery_app

# 建立任務函數
@celery_app.task
def my_task1():
    print("任務函數(my_task1)正在執行....")

@celery_app.task
def my_task2():
    print("任務函數(my_task2)正在執行....")

@celery_app.task
def my_task3():
    print("任務函數(my_task3)正在執行....")
  啓動worker:

celery -A proj worker -l info


鍵入ctrl+c可關閉worker.

3.2調度任務

調用任務，可以使用delay()方法:

my_task.delay(2, 2)
  也可使用apply_async()方法，該方法可以讓咱們設置一些任務執行的參數，例如，任務多久以後才執行，任務被髮送到那個隊列中等等.

my_task.apply_async((2, 2), queue='my_queue', countdown=10)
任務my_task將會被髮送到my_queue隊列中，而且在發送10秒以後執行。

  若是咱們直接執行任務函數，將會直接執行此函數在當前進程中，並不會向broker發送任何消息。

  不管是delay()仍是apply_async()方式都會返回AsyncResult對象，方便跟蹤任務執行狀態，但須要咱們配置result_backend.

  每個被吊用的任務都會被分配一個ID，咱們叫Task ID.

3.3Desining Work-flows

signature

咱們到目前爲止只是知道如何使用delay()方法，固然這個方法也是很是經常使用的。可是有時咱們並不想簡單的將任務發送到隊列中，咱們想將一個任務函數(由參數和執行選項組成)做爲一個參數傳遞給另一個函數中，爲了實現此目標，Celery使用一種叫作signatures的東西。

一個signature包裝了一個參數和執行選項的單個任務調用。咱們可將這個signature傳遞給函數。

咱們先看下tasks.py模塊中定義的任務函數:

from proj.celery import app as celery_app # 建立任務函數 @celery_app.task def my_task1(): print("任務函數(my_task1)正在執行....") @celery_app.task def my_task2(): print("任務函數(my_task2)正在執行....") @celery_app.task def my_task3(): print("任務函數(my_task3)正在執行....")

咱們將my_task1()任務包裝稱一個signature:

t1 = my_task1.signatures(countdown=10) t1.delay()

Primitives

這些primitives自己就是signature對象，所以它們能夠以多種方式組合成複雜的工做流程。primitives以下:

group: 一組任務並行執行，返回一組返回值，並能夠按順序檢索返回值。

chain: 任務一個一個執行，一個執行完將執行return結果傳遞給下一個任務函數.

tasks.py模塊以下:

from proj.celery import app as celery_app # 建立任務函數 @celery_app.task def my_task1(a, b): print("任務函數(my_task1)正在執行....") return a + b @celery_app.task def my_task2(a, b): print("任務函數(my_task2)正在執行....") return a + b @celery_app.task def my_task3(a, b): print("任務函數(my_task3)正在執行....") return a + b

group案例以下(test.py模塊):

from proj.tasks import my_task1 from proj.tasks import my_task2 from proj.tasks import my_task3 from celery import group # 將多個signature放入同一組中 my_group = group((my_task1.s(10, 10), my_task2.s(20, 20), my_task3.s(30, 30))) ret = my_group() # 執行組任務 print(ret.get()) # 輸出每一個任務結果

chain案例以下(test.py模塊):

from proj.tasks import my_task1 from proj.tasks import my_task2 from proj.tasks import my_task3 from celery import chain # 將多個signature組成一個任務鏈 # my_task1的運行結果將會傳遞給my_task2 # my_task2的運行結果會傳遞給my_task3 my_chain = chain(my_task1.s(10, 10) | my_task2.s(20) | my_task3.s(30)) ret = my_chain() # 執行任務鏈 print(ret.get()) # 輸出最終結果

3.4Routing

 假如咱們有兩個worker,一個worker專門用來處理郵件發送任務和圖像處理任務，一個worker專門用來處理文件上傳任務。

  咱們建立兩個隊列，一個專門用於存儲郵件任務隊列和圖像處理，一個用來存儲文件上傳任務隊列。

  Celery支持AMQP(Advanced Message Queue)全部的路由功能，咱們也可使用簡單的路由設置將指定的任務發送到指定的隊列中.

  咱們須要配置在celeryconfig.py模塊中配置 CELERY_ROUTES 項, tasks.py模塊修改以下:

from proj.celery import app as celery_app


@celery_app.task
def my_task1(a, b):
    print("my_task1任務正在執行....")
    return a + b


@celery_app.task
def my_task2(a, b):
    print("my_task2任務正在執行....")
    return a + b


@celery_app.task
def my_task3(a, b):
    print("my_task3任務正在執行....")
    return a + b


@celery_app.task
def my_task4(a, b):
    print("my_task3任務正在執行....")
    return a + b


@celery_app.task
def my_task5():
    print("my_task5任務正在執行....")


@celery_app.task
def my_task6():
    print("my_task6任務正在執行....")


@celery_app.task
def my_task7():
    print("my_task7任務正在執行....")
  咱們經過配置，將send_email和upload_file任務發送到queue1隊列中，將image_process發送到queue2隊列中。

  咱們修改celeryconfig.py:

broker_url='redis://:@127.0.0.1:6379/1'
result_backend='redis://:@127.0.0.1:6379/2'


task_routes=({
    'proj.tasks.my_task5': {'queue': 'queue1'},
    'proj.tasks.my_task6': {'queue': 'queue1'},
    'proj.tasks.my_task7': {'queue': 'queue2'},
    },
)
  test.py:

from proj.tasks import *

# 發送任務到路由指定的隊列中
my_task5.delay() my_task6.delay() my_task7.delay()

&emsp;&emsp;開啓兩個worker服務器，分別處理兩個隊列：
```python
celery -A proj worker --loglevel=info -Q queue1
celery -A proj worker --loglevel=info -Q queue2
  咱們一樣也能夠經過apply_aynsc()方法來設置任務發送到那個隊列中:

my_task1.apply_async(queue='queue1')
  咱們也可設置一個worker服務器處理兩個隊列中的任務:

celery -A proj worker --loglevel=info -Q queue1,queue2

4.Periodic Tasks(週期性任務)

celery beat是一個調度器，它能夠週期內指定某個worker來執行某個任務。若是咱們想週期執行某個任務須要增長beat_schedule配置信息.  

broker_url='redis://:@127.0.0.1:6379/1'
result_backend='redis://:@127.0.0.1:6379/2'

# 指定任務發到那個隊列中
task_routes=({
    'proj.tasks.my_task5': {'queue': 'queue1'},
    'proj.tasks.my_task6': {'queue': 'queue1'},
    'proj.tasks.my_task7': {'queue': 'queue2'},
    },
)


# 配置週期性任務，　或者定時任務
beat_schedule = {
    'every-5-seconds':
        {
            'task': 'proj.tasks.my_task8',
            'schedule': 5.0,
            # 'args': (16, 16),
        }
}
  tasks.py模塊內容以下:

from proj.celery import app as celery_app


@celery_app.task
def my_task1(a, b):
    print("my_task1任務正在執行....")
    return a + b


@celery_app.task
def my_task2(a, b):
    print("my_task2任務正在執行....")
    return a + b


@celery_app.task
def my_task3(a, b):
    print("my_task3任務正在執行....")
    return a + b


@celery_app.task
def my_task4(a, b):
    print("my_task3任務正在執行....")
    return a + b


@celery_app.task
def my_task5():
    print("my_task5任務正在執行....")




@celery_app.task
def my_task6():
    print("my_task6任務正在執行....")



@celery_app.task
def my_task7():
    print("my_task7任務正在執行....")


# 週期執行任務
@celery_app.task
def my_task8():
    print("my_task8任務正在執行....")
  啓動woker處理週期性任務:

celery -A proj worker --loglevel=info --beat
  若是咱們想指定在某天某時某分某秒執行某個任務，能夠執行cron任務, 增長配置信息以下:

beat_schedule = {
    'every-5-minute':
        {
            'task': 'proj.tasks.period_task',
            'schedule': 5.0,
            'args': (16, 16),
        },
    'add-every-monday-morning': {
        'task': 'proj.tasks.period_task',
        'schedule': crontab(hour=7, minute=30, day_of_week=1),
        'args': (16, 16),
    },

}
crontab例子: http://docs.celeryproject.org/en/latest/userguide/periodic-tasks.html

  開啓一個celery beat服務：

celery -A proj beat
  celery須要保存上次任務運行的時間在數據文件中，文件在當前目錄下名字叫celerybeat-schedule. beat須要訪問此文件：

celery -A proj beat -s /home/celery/var/run/celerybeat-schedule

5.Django使用Celery

5.1配置celery

1. 配置celery
  建立django項目celery_demo, 建立應用demo:

django-admin startproject celery_demo
python manage.py startapp demo



  在celery_demo模塊中建立celery.py模塊, 文件目錄爲:



   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)
  在應用demo引用建立tasks.py模塊, 文件目錄爲:



  咱們在文件內建立一個任務函數my_task:

from celery_demo.celery import app
import time

# 加上app對象的task裝飾器
# 此函數爲任務函數
@app.task
def my_task():
    print("任務開始執行....")
    time.sleep(5)
    print("任務執行結束....")
  在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>")
  在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),
]
  建立worker等待處理celery隊列中任務, 在終端執行命令:

celery -A celery_demo worker -l info


  啓動django測試服務器：

python manage.py runserver

5.2保存任務結果

此處須要用到額外包django_celery_results, 先安裝包:

pip install django-celery-results
  在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',  # 注意此處應用名爲下劃線
]
  回到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)
  建立django_celery_results應用所需數據庫表, 執行遷移文件：

python manage.py migrate django_celery_results
  我這裏使用的是django默認的數據庫sqlit, 執行遷移以後，會在數據庫中建立一張用來存儲任務結果的表:


  再次從瀏覽器發送請求， 任務執行結束以後，將任務結果保存在數據庫中:

5.3定時任務

若是咱們想某日某時執行某個任務，或者每隔一段時間執行某個任務，也可使用celery來完成.   使用定時任務，須要安裝額外包:

pip install django_celery_beat
  首先在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',  # 安裝應用
]
  在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)
  因爲定時器信息存儲在數據庫中，咱們須要先生成對應表, 對diango_celery_beat執行遷移操做，建立對應表:

python manage.py migrate django_celery_beat




  咱們可登陸網站後臺Admin去建立對應任務, 首先咱們先在tasks.py模塊中增長新的任務，用於定時去執行(5秒執行一次)

from celery_demo.celery import app
import time

# 用於定時執行的任務
@app.task
def interval_task():
    print("我每隔5秒鐘時間執行一次....")
  首先建立後臺管理員賬號:

python manage.py createsuperuser
  登陸管理後臺Admin:



  其中Crontabs用於定時某個具體時間執行某個任務的時間，Intervals用於每隔多久執行任務的事件，具體任務的執行在Periodic tasks表中建立。

  咱們要建立每隔5秒執行某個任務，因此在Intervals表名後面點擊Add按鈕:




  而後在Periodic tasks表名後面，點擊Add按鈕，添加任務:



  啓動定時任務：

celery -A celery_demo worker -l info --beat



  任務每隔5秒中就會執行一次，若是配置了存儲，那麼每次任務執行的結果也會被保存到對應的數據庫中。