python測試開發django-159.Celery 異步與 RabbitMQ 環境搭建

前言

Celery是一個Python任務隊列系統，用於處理跨線程或網絡節點的工作任務分配。它使異步任務管理變得容易。
您的應用程序只需要將消息推送到像RabbitMQ這樣的代理，Celery worker會彈出它們並安排任務執行。

Celery

celery 的5個角色

• Task 就是任務，有異步任務(Async Task)和定時任務(Celery Beat)
• Broker 中間人，接收生產者發來的消息即Task，將任務存入隊列。任務的消費者是Worker。Celery 本身不提供隊列服務，推薦用Redis或RabbitMQ實現隊列服務。
• Worker 執行任務的單元，它實時監控消息隊列，如果有任務就獲取任務並執行它。
• Beat 定時任務調度器，根據配置定時將任務發送給Broker。
• Backend 用於存儲任務的執行結果。

環境准備

1.django環境v2.1.2
2.安裝celery版本

pip install celery==3.1.26.post2

3.安裝django-celery包

pip install django-celery==3.3.1

RabbitMQ 環境

Broker(RabbitMQ) 負責創建任務隊列，根據一些路由規則將任務分派到任務隊列，然后將任務從任務隊列交付給 worker
先使用docker 搭建RabbitMQ 環境,rabbitMQ 鏡像倉庫地址 https://hub.docker.com/_/rabbitmq找帶有 mangement的版本，會帶web后台管理界面

下載 3.8.0-management 鏡像

docker pull rabbitmq:3.8.0-management

啟動容器,設置賬號 admin 和密碼 123456

docker run -d --name rabbitmq3.8 -p 5672:5672 -p 15672:15672 --hostname myRabbit -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=123456 rabbitmq:3.8.0-management

宿主機需開放 5672 和 15672 這 2 個端口，5672 是后端接口訪問的端口，15672 是前端 web 管理后台頁面地址，輸入http://ip:15672可以訪問 web 網站

輸入前面設置的賬號 admin 和密碼 123456 可以直接登錄

Django 中使用 Celery

要在 Django 項目中使用 Celery，您必須首先定義 Celery 庫的一個實例（稱為“應用程序”）

如果你有一個現代的 Django 項目布局，比如：

- proj/
  - manage.py
  - proj/
    - __init__.py
    - settings.py
    - urls.py

那么推薦的方法是創建一個新的proj/proj/celery.py模塊來定義 Celery 實例：

import os

from celery import Celery

# Set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings')

app = Celery('proj')

# Using a string here means the worker doesn't have to serialize
# the configuration object to child processes.
# - namespace='CELERY' means all celery-related configuration keys
#   should have a `CELERY_` prefix.
app.config_from_object('django.conf:settings', namespace='CELERY')

# Load task modules from all registered Django apps.
app.autodiscover_tasks()


@app.task(bind=True)
def debug_task(self):
    print(f'Request: {self.request!r}')

其中debug_task是測試的任務，可以注掉

# @app.task(bind=True)
# def debug_task(self):
#     print('Request: {0!r}'.format(self.request))

上面一段只需改這句，'proj'是自己django項目的app名稱

app = Celery('proj')

然后你需要在你的proj/proj/__init__.py 模塊中導入這個應用程序。這確保在 Django 啟動時加載應用程序，以便@shared_task裝飾器（稍后提到）將使用它：

# 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',)

上面這段固定的，不用改

tasks任務

在app下新建tasks.py,必須要是tasks.py文件名稱，django會自動查找到app下的該文件

@shared_task
def add(x, y):
    print("task----------1111111111111111111111")
    return x + y


@shared_task
def mul(x, y):
    return x * y

tasks.py可以寫任務函數add、mul，讓它生效的最直接的方法就是添加app.task 或shared_task 這個裝飾器

添加setting配置

setting.py添加配置

• broker參數表示用來連接broker的URL，rabbitmq采用的是一種稱為’amqp’的協議，如果rabbitmq運行在默認設置下，celery不需要其他信息，只要amqp://即可。
• backend參數是可選的，如果想要查詢任務狀態或者任務執行結果時必填, Celery中的后端用於存儲任務結果。rpc意味着將結果作為AMQP消息發送回去。

#   RabbitMQ配置BROKER_URL 和backend
BROKER_URL = 'amqp://admin:123456@192.168.1.11:5672//'
CELERY_RESULT_BACKEND = 'rpc://'

CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_ACCEPT_CONTENT=['json']
CELERY_TIMEZONE = 'Asia/Shanghai'
CELERY_ENABLE_UTC = True

創建視圖

views.py創建視圖

from .tasks import add, mul

def task_demo(request):
    res = add.delay(10, 20)
    print(res.task_id)  # 返回task_id
    return JsonResponse({"code": 0, "res": res.task_id})

啟動worker

前面pip已經安裝過celery應用了，celery是一個獨立的應用，可以啟動worker

celery -A MyDjango worker -l info

其中MyDjango是你自己的django項目名稱

運行日志

 -------------- celery@DESKTOP-HJ487C8 v3.1.26.post2 (Cipater)
---- **** -----
--- * ***  * -- Windows-10-10.0.17134-SP0
-- * - **** ---
- ** ---------- [config]
- ** ---------- .> app:         yoyo:0x1ea1a96e9b0
- ** ---------- .> transport:   amqp://admin:**@192.168.1.11:5672//
- ** ---------- .> results:     rpc://
- *** --- * --- .> concurrency: 4 (prefork)
-- ******* ----
--- ***** ----- [queues]
 -------------- .> celery           exchange=celery(direct) key=celery


[tasks]
  . yoyo.tasks.add
  . yoyo.tasks.mul

[2021-10-18 22:45:03,155: INFO/MainProcess] Connected to amqp://admin:**@192.168.1.11:5672//
[2021-10-18 22:45:03,347: INFO/MainProcess] mingle: searching for neighbors
[2021-10-18 22:45:04,897: INFO/MainProcess] mingle: all alone
[2021-10-18 22:45:05,406: WARNING/MainProcess] e:\python36\lib\site-packages\celery\fixups\django.py:265: 
UserWarning: Using settings.DEBUG leads to a memory leak, never use this setting in production environments!
  warnings.warn('Using settings.DEBUG leads to a memory leak, never '
[2021-10-18 22:45:05,407: WARNING/MainProcess] celery@DESKTOP-HJ487C8 ready.

運行的時候，當我們看到" Connected to amqp"說明已經連接成功了！

shell交互環境

在django shell交互環境調試運行任務

D:\202107django\MyDjango>python manage.py shell
Python 3.6.6 (v3.6.6:4cf1f54eb7, Jun 27 2018, 03:37:03) [MSC v.1900 64 bit (AMD64)] on win32
Type "help", "copyright", "credits" or "license" for more information.
(InteractiveConsole)
>>> from yoyo.tasks import add,mul
>>> from celery.result import AsyncResult
>>>
>>> res = add.delay(11, 12)
>>> res
<AsyncResult: c5ff83a4-4840-4b36-8869-5ce6081904f1>
>>> res.status
'SUCCESS'
>>>
>>> res.backend
<celery.backends.redis.RedisBackend object at 0x0000015E011C3128>
>>>
>>> res.task_id
'c5ff83a4-4840-4b36-8869-5ce6081904f1'
>>>
>>>
>>> get_task = AsyncResult(id=res.task_id)
>>> get_task
<AsyncResult: c5ff83a4-4840-4b36-8869-5ce6081904f1>
>>> get_task.result
23
>>>

res.status是查看任務狀態
res.task_id 是獲取任務的id
res.result 獲取任務結果
根據任務的id查詢任務的執行結果AsyncResult(id=res.task_id).result獲取