python3+celery+redis實現異步任務

1、原理javascript

Celery是基於Python開發的一個分佈式任務隊列框架，支持使用任務隊列的方式在分佈的機器/進程/線程上執行任務調度。它是Python寫的庫，可是它實現的通信協議也能夠使用ruby，php，javascript等調用。異步任務除了消息隊列的後臺執行的方式，仍是一種則是定時計劃任務。php

Celery 是一個強大的分佈式任務隊列，它可讓任務的執行徹底脫離主程序，甚至能夠被分配到其餘主機上運行。咱們一般使用它來實現異步任務（async task）和定時任務（crontab）。它的架構組成以下圖 java

組件：python

一、任務（tasks）--用戶定義的函數，用於實現用戶的功能，好比執行一個耗時很長的任務web

二、中間介（Broker）--用於存放tasks的地方，可是這個中間介須要解決一個問題，就是可能須要存放很是很是多的tasks，並且要保證Worker可以從這裏拿取redis

三、執行者（Worker）--用於執行tasks，也就是真正調用咱們在tasks中定義的函數數據庫

四、存儲（Backend）--把執行tasks返回的結果進行存儲，以供用戶查看或調用django

2、實現過程json

1.環境安裝（RabbitMQ/Redis、Celery、django-celery、flower）ruby

個人python版本：3.7，celery版本：3.1.26.post2

2.建立工程

紅圈爲本工程所需：

web_order下面須要修改的文件：celery.py、__init__.py、settings文件

web_test下面須要修改的文件：tasks.py文件、longTask.py文件

3.修改文件過程

1）修改settings.py。在settings的最後加上以下代碼：

1 # CELERY SETTING
2 BROKER_URL = 'redis://localhost:6379'    #指定消息中間件
3 CELERY_RESULT_BACKEND = 'redis://localhost:6379'   ##指定結果存儲位置爲本地數據庫
4 CELERY_ACCEPT_CONTENT = ['application/json']  # 5 CELERY_TASK_SERIALIZER = 'json'
6 CELERY_RESULT_SERIALIZER = 'json'
7 CELERY_TIMEZONE = 'Asia/Shanghai'
8 CELERY_IMPORTS = ("web_test.tasks")   #註冊任務，

###在 INSTALLED_APPS中註冊 djcelery

2）__init__ 文件。

#絕對導入，以避免celery和標準庫中的celery模塊衝突
from __future__ import absolute_import # This will make sure the app is always imported when # Django starts so that shared_task will use this app.

#如下導入時爲了確保在Django啓動時加載app，shared_task在app中會使用到
from .celery import app as celery_app __all__ = ['celery_app']

3）celery文件

from __future__ import absolute_import,unicode_literals import os from celery import Celery from django.conf import settings os.environ.setdefault("DJANGO_SETTINGS_MODULE", "web_order.settings")  # 設置celery能夠在命令行中使用
app = Celery('web_order')  # 建立app實例 # app = Celery('tcelery', backend='redis://localhost:6379/0', broker='redis://localhost:6379/0')
app.conf.CELERY_IGNORE_RESULT = False  # 結果不忽略 # app.conf.CELERY_RESULT_BACKEND = 'redis://localhost:6379/0' #結果保存在redis中
 app.config_from_object('django.conf:settings')  # 從文件中加載實例
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)  # 自動加載tasks，注意：他會去app下面查找tasks.py文件，因此咱們必須將task放在tasks.py文件中

4）longTasks.py

from django.http import JsonResponse import json from .tasks import test   #導入異步任務的方法


def sendlongtask(request): #由此處調用test方法，執行異步命令
    run_res = test(sh, userIp, username) return JsonResponse("執行成功", safe=False)

5）Tasks.py

from celery import shared_task @shared_task def test(x, y): return (x+y)

4.啓動Django和celery

　　在項目根目錄執行：

　　python manage.py runserver 0.0.0.0:8000

　　python manage.py celery worker -c 4 --loglevel=info

5.另外，Celery提供了一個工具flower，將各個任務的執行狀況、各個worker的健康狀態進行監控並以可視化的方式展示，以下圖所示：

　　　　Django下實現的方式以下：　

　　　　　　1.）安裝flower:

　　　　pip install flower

　　　　　　2.）啓動flower（默認會啓動一個webserver，端口爲5555）:

python manage.py celery flower

　　　　　3.）進入http://localhost:5555便可查看。

6.python3踩過的坑：python3.7與celery不兼容

1.出現這個錯誤時，須要將報錯文件中全部的async改成asynchronous或者其餘變量名便可。

運行後，你會發現celery能夠正常使用了

2.將python版本降到3.6及如下，celery也可正常使用。