摘要：今天介紹一下如何在項(xiàng)目中使用搭建一個(gè)有兩個(gè)節(jié)點(diǎn)的任務(wù)隊(duì)列一個(gè)主節(jié)點(diǎn)一個(gè)子節(jié)點(diǎn)主節(jié)點(diǎn)發(fā)布任務(wù)，子節(jié)點(diǎn)收到任務(wù)并執(zhí)行。

今天介紹一下如何在django項(xiàng)目中使用celery搭建一個(gè)有兩個(gè)節(jié)點(diǎn)的任務(wù)隊(duì)列(一個(gè)主節(jié)點(diǎn)一個(gè)子節(jié)點(diǎn)；主節(jié)點(diǎn)發(fā)布任務(wù)，子節(jié)點(diǎn)收到任務(wù)并執(zhí)行。搭建3個(gè)或者以上的節(jié)點(diǎn)就類似了)，使用到了celery,rabbitmq。這里不會(huì)多帶帶介紹celery和rabbitmq中的知識(shí)了。

1.項(xiàng)目基礎(chǔ)環(huán)境：
兩個(gè)ubuntu18.04虛擬機(jī)、python3.6.5、django2.0.4、celery3.1.26post2

2.主節(jié)點(diǎn)django項(xiàng)目結(jié)構(gòu)：

3.settings.py中關(guān)于celery的配置：

import djcelery
# 此處的Queue和Exchange都涉及到RabbitMQ中的概念，這里不做介紹
from kombu import Queue, Exchange
djcelery.setup_loader()
BROKER_URL = "amqp://test:[email protected]:5672/testhost"
CELERY_RESULT_BACKEND = "amqp://test:[email protected]:5672/testhost"

CELERY_TASK_RESULT_EXPIRES=3600
CELERY_TASK_SERIALIZER="json"
CELERY_RESULT_SERIALIZER="json"
# CELERY_ACCEPT_CONTENT = ["json", "pickle", "msgpack", "yaml"]

CELERY_DEFAULT_EXCHANGE = "train"
CELERY_DEFAULT_EXCHANGE_TYPE = "direct"

CELERY_IMPORTS = ("proj.celery1.tasks", )

CELERY_QUEUES = (
  Queue("train", routing_key="train"),
  Queue("predict", routing_key="predict"),
)

4.celery.py中的配置：

# coding:utf8
from __future__ import absolute_import

import os

from celery import Celery
from django.conf import settings

# 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 will not have to
# pickle the object when using Windows.
app.config_from_object("django.conf:settings")
# app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
app.autodiscover_tasks(settings.INSTALLED_APPS)


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

5.proj/init.py中的配置：

from __future__ import absolute_import
from .celery import app as celery_app

6.celery1/tasks.py：（主節(jié)點(diǎn)中的任務(wù)不會(huì)執(zhí)行，只執(zhí)行子節(jié)點(diǎn)中的任務(wù)）

from __future__ import absolute_import
from celery import task


@task
def do_train(x, y):
    return x + y

7.celery1/views.py：

from .tasks import do_train
class Test1View(APIView):
    def get(self, request):
        try:
            # 這里的queue和routing_key也涉及到RabiitMQ中的知識(shí)
            # 關(guān)鍵，在這里控制向哪個(gè)queue中發(fā)送任務(wù)，子節(jié)點(diǎn)通過(guò)這個(gè)執(zhí)行對(duì)應(yīng)queue中的任務(wù)
            ret = do_train.apply_async(args=[4, 2], queue="train", routing_key="train")
            # 獲取結(jié)果
            data = ret.get()
        except Exception as e:
            return Response(dict(msg=str(e), code=10001))
        return Response(dict(msg="OK", code=10000, data=data))

8.子節(jié)點(diǎn)目錄結(jié)構(gòu)：

9.子節(jié)點(diǎn)中celery1/celery.py：

from __future__ import absolute_import
from celery import Celery
CELERY_IMPORTS = ("celery1.tasks", )
app = Celery("myapp",
             # 此處涉及到RabbitMQ的知識(shí)，RabbitMQ是對(duì)應(yīng)主節(jié)點(diǎn)上的
             broker="amqp://test:[email protected]:5672/testhost",
             backend="amqp://test:[email protected]:5672/testhost",
             include=["celery1.tasks"])

app.config_from_object("celery1.config")

if __name__ == "__main__":
  app.start()

10.子節(jié)點(diǎn)中celery1/config.py：

from __future__ import absolute_import
from kombu import Queue,Exchange
from datetime import timedelta

CELERY_TASK_RESULT_EXPIRES=3600
CELERY_TASK_SERIALIZER="json"
CELERY_RESULT_SERIALIZER="json"
CELERY_ACCEPT_CONTENT = ["json","pickle","msgpack","yaml"]

CELERY_DEFAULT_EXCHANGE = "train"
# exchange type可以看RabbitMQ中的相關(guān)內(nèi)容
CELERY_DEFAULT_EXCHANGE_TYPE = "direct"

CELERT_QUEUES =  (
  Queue("train",exchange="train",routing_key="train"),
)

11.子節(jié)點(diǎn)celery1/tasks.py：(這個(gè)是要真正執(zhí)行的task，每個(gè)節(jié)點(diǎn)可以不同)

from __future__ import absolute_import
from celery1.celery import app


import time
from celery import task


@task
def do_train(x, y):
    """
    訓(xùn)練
    :param data:
    :return:
    """
    time.sleep(3)
    return dict(data=str(x+y),msg="train")

12.啟動(dòng)子節(jié)點(diǎn)中的celery:
celery1是項(xiàng)目，-Q train表示從train這個(gè)queue中接收任務(wù)

celery -A celery1 worker -l info -Q train

13.啟動(dòng)主節(jié)點(diǎn)中的django項(xiàng)目:

python manage.py runserver

14.使用Postman請(qǐng)求對(duì)應(yīng)的view

請(qǐng)求url：http://127.0.0.1:8000/api/v1/celery1/test/
返回的結(jié)果是:
{
    "msg": "OK",
    "code": 10000,
    "data": {
        "data": "6",
        "msg": "train"
    }
}

15.遇到的問(wèn)題：
1）celery隊(duì)列報(bào)錯(cuò): AttributeError: ‘str’ object has no attribute ‘items’
解決：將redis庫(kù)從3.0回退到了2.10，pip install redis==2.10
解決方法參考鏈接：https://stackoverflow.com/que...

今天就說(shuō)到這里，如有疑問(wèn)，歡迎交流。