Celery 異步任務、定時任務

時間  2019-11-12
標籤 celery 異步 任務 定時 简体版
原文   原文鏈接

celery介紹

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

  • 簡單，易於使用和維護，有豐富的文檔。
  • 高效，單個celery進程每分鐘能夠處理數百萬個任務。
  • 靈活，celery中幾乎每一個部分均可以自定義擴展。

celery很是易於集成到一些web開發框架中.python

Task Queue

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

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

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

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

 Celery 安裝

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

pip install -U Celery

也可從官方直接下載安裝包:https://pypi.python.org/pypi/celery/json

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

 Broker

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

1.RabbitMQ

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

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

2.Redis

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

Application

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

1.建立應用

咱們首先建立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

顯示效果以下: 

2.調用任務

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

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

from tasks import my_task
my_task.delay()

執行效果以下: 

 

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

3.存儲結果

  若是咱們想跟蹤任務的狀態，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

 配置

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

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

1.直接經過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',
)

2.專有配置文件

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

  配置模塊的名字咱們取名爲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')

更多配置: http://docs.celeryproject.org/en/latest/userguide/configuration.html#configuration

 在項目中使用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

 

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, 執行遷移以後，會在數據庫中建立一張用來存儲任務結果的表: 

 

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秒中就會執行一次，若是配置了存儲，那麼每次任務執行的結果也會被保存到對應的數據庫中。

調用tasks.py的方法時，有時會出現傳入的參數沒法被序列表化，解決方法：將傳入的參數以字典的形式包裝成一個字典，確保字典可以被json序列化，及key和value都是能被json序列化的。

在centos上使用pip或者編譯安裝了celery卻提示找不到celery這個命令時，解決方法：

export PATH=/usr/local/python3/bin:$PATH      # 個人Python3資源文件都放在 /usr/local/python3 文件夾下

echo 'export PATH=/usr/local/python3/bin:$PATH' >> /etc/profile.d/python3.sh  # 避免重啓丟失

若是提示沒有權限 就加sudo

在windows10上使用celery，使用celery -A celery_demo worker -l info命令出錯,解決方法：安裝協程模塊pip install eventlet， 執行命令celery -A celery_demo worker -l info -P eventlet

在windows10上使用celery時，異步任務的啓動命令不能和定時任務的不能一塊兒啓動，要分開使用

celery -A celery_demo worker -l info -P eventlet

celery -A celery_demo beat -l info

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程