我有celery 片段如下：
首先在celery_tasks.tasks.py中定义任务类

import celery

class LoggerDefine(celery.Task):
    name = 'message-logger'

    def run(self, payload):
        pass

class PerformanceMeasureDefine(celery.Task):
    name = 'performance-logger'

    def run(self, payload, elapseTime):):
        pass

字符串
然后从consumers.logger中实现方法

import django
django.setup()
from log_models.models import Level_logs
from django.conf import settings
from common_functions.logger import *
from celery_tasks.tasks import LoggerDefine, PerformanceMeasureDefine


class LoggerImpl(LoggerDefine):
    def run(self, payload):
        fmt1 = 'Message body: {}'
        print("-------------------------------- log persistor consuming --------------------------------")
        Level_logs.objects.create(
            level=payload.get(LEVEL),
            date_time=payload.get(DATETIME),
            file_name=payload.get(FILENAME),
            line_number=payload.get(LINE_NUMBER),
            function_name=payload.get(FUNCTION),
            payload=payload.get(PAYLOAD),
        )
        return fmt1.format(payload)



class PerformanceMeasureImpl(PerformanceMeasureDefine):
    def run(self, payload, elapseTime):
        fmt1 = 'Message body: {}'
        print("-------------------------------- performance consuming --------------------------------")
        print("payload = ", payload)
        print("elapseTime = ", elapseTime)
        return fmt1.format(payload)

型
在celery_tasks.utils中创建celery工作者和消费者的函数

import celery

def create_worker_from(WorkerClass):
    assert issubclass(WorkerClass, celery.Task)
    app = celery.Celery()
    app.config_from_object("django.conf:settings")
    app.conf.update(task_default_queue=WorkerClass.name) # update worker queue
    worker_task = app.register_task(WorkerClass())

    return app, worker_task

型
其中django setting.py文件中的celery设置是：

CELERY_BROKER_URL = "amqp://guest:guest@rabbitmq:5672//"

型
最后，我将它们示例化为两个消费者

from consumers.logger import LoggerImpl, PerformanceMeasureImpl
from celery_tasks.utils import create_worker_from

logger, _ = create_worker_from(LoggerImpl)
logger.worker_main(["worker", "-c", "2"])

performance_measure, _ = create_worker_from(PerformanceMeasureImpl)
performance_measure.worker_main(["worker", "-c", "2"])

型
然后它不工作，我试图注解掉performance_measure并只留下logger如下：

from consumers.logger import LoggerImpl, PerformanceMeasureImpl
from celery_tasks.utils import create_worker_from

logger, _ = create_worker_from(LoggerImpl)
logger.worker_main(["worker", "-c", "2"])

#performance_measure, _ = create_worker_from(PerformanceMeasureImpl)
#performance_measure.worker_main(["worker", "-c", "2"])

型
然后logger工作正常。我还尝试注解掉logger，只留下performance_measure，如下所示：

from consumers.logger import LoggerImpl, PerformanceMeasureImpl
from celery_tasks.utils import create_worker_from

#logger, _ = create_worker_from(LoggerImpl)
#logger.worker_main(["worker", "-c", "2"])

performance_measure, _ = create_worker_from(PerformanceMeasureImpl)
performance_measure.worker_main(["worker", "-c", "2"])

型
那么performance_measure工作正常。

因此，只有一个消费者正常工作，但两个消费者在一起就不正常了。

当两个消息同时工作时，有一个错误和消息卡在消息队列中而没有消耗。

consumer: Cannot connect to amqp://guest:**@rabbitmq:5672//: [Errno 111] Connection refused.

型
顺便说一句，发布者的用法就像

_, logger_worker = create_worker_from(LoggerDefine)
_, performance_measure_worker = create_worker_from(PerformanceMeasureDefine)

logger_worker.apply_async(args=[msg,])
performance_measure_worker.apply_async(
                args=[request.data, elapsed_time])

型
我怎么能让其中两个工作？
多谢了。