我有一个使用Django的Web应用程序，我正在使用Celery进行一些异步任务处理。
对于Celery，我使用Rabbitmq作为代理，并使用Redis作为结果后端。
Rabbitmq和Redis运行在同一台Ubuntu 14.04服务器上，托管在本地虚拟机上。
Celery工作者正在远程计算机（Windows 10）上运行（没有工作者在Django服务器上运行）。
我有三个问题（我认为它们某种程度上是相关的！）：
- 无论任务是否成功或失败，任务都保持“PENDING”状态。
- 任务失败时不会重试。尝试重试时，我收到以下错误：
reject requeue=False: [WinError 10061] No connection could be made because the target machine actively refused it
- 结果后端似乎没起作用。
我对我的设置也感到困惑，不确定这些问题确切来自哪里！
因此，这是我的设置：
my_app/settings.py
# region Celery Settings
CELERY_CONCURRENCY = 1
CELERY_ACCEPT_CONTENT = ['json']
# CELERY_RESULT_BACKEND = 'redis://:C@pV@lue2016@cvc.ma:6379/0'
BROKER_URL = 'amqp://soufiaane:C@pV@lue2016@cvc.ma:5672/cvcHost'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TASK_SERIALIZER = 'json'
CELERY_ACKS_LATE = True
CELERYD_PREFETCH_MULTIPLIER = 1
CELERY_REDIS_HOST = 'cvc.ma'
CELERY_REDIS_PORT = 6379
CELERY_REDIS_DB = 0
CELERY_RESULT_BACKEND = 'redis'
CELERY_RESULT_PASSWORD = "C@pV@lue2016"
REDIS_CONNECT_RETRY = True
AMQP_SERVER = "cvc.ma"
AMQP_PORT = 5672
AMQP_USER = "soufiaane"
AMQP_PASSWORD = "C@pV@lue2016"
AMQP_VHOST = "/cvcHost"
CELERYD_HIJACK_ROOT_LOGGER = True
CELERY_HIJACK_ROOT_LOGGER = True
CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler'
# endregion
我的应用程序/celery_settings.py
from __future__ import absolute_import
from django.conf import settings
from celery import Celery
import django
import os
# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'my_app.settings')
django.setup()
app = Celery('CapValue', broker='amqp://soufiaane:C@pV@lue2016@cvc.ma/cvcHost', backend='redis://:C@pV@lue2016@cvc.ma:6379/0')
# 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.task(bind=True)
def debug_task(self):
print('Request: {0!r}'.format(self.request))
my_app__init__.py
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.
from .celery_settings import app as celery_app
我的应用程序\电子邮件\tasks.py
from __future__ import absolute_import
from my_app.celery_settings import app
# here i only define the task skeleton because i'm executing this task on remote workers !
@app.task(name='email_task', bind=True, max_retries=3, default_retry_delay=1)
def email_task(self, job, email):
try:
print("x")
except Exception as exc:
self.retry(exc=exc)
在工人方面，我有一个名为“tasks.py”的文件，其中包含任务的实际实现：
工人\tasks.py
from __future__ import absolute_import
from celery.utils.log import get_task_logger
from celery import Celery
logger = get_task_logger(__name__)
app = Celery('CapValue', broker='amqp://soufiaane:C@pV@lue2016@cvc.ma/cvcHost', backend='redis://:C@pV@lue2016@cvc.ma:6379/0')
@app.task(name='email_task', bind=True, max_retries=3, default_retry_delay=1)
def email_task(self, job, email):
try:
"""
The actual implementation of the task
"""
except Exception as exc:
self.retry(exc=exc)
然而我注意到的是：
- 当我将工作进程中的代理设置更改为错误密码时，会出现无法连接到代理的错误。
- 当我将工作进程中结果后端的设置更改为错误密码时，它会正常运行，好像一切都没问题。
什么可能导致这些问题？
编辑
在我的Redis服务器上，我已经启用了远程连接
/etc/redis/redis.conf
... bind 0.0.0.0 ...
my_app/settings.py和my_app/celery_settings.py中重复了Celery配置，并且你没有一个celeryconfig.py（这在独立的Celery中很常见）- 这是推荐的做法吗？你使用的是哪个文档？ - scytalesudo rabbitmqctl list_queues -p cvcHost。我认为结果后端似乎没有工作，因为你的任务没有完成。 - Adi Krishnan