django 和 celery+rabbitmq 的消费者连接错误

使用 Django 和 Celery+RabbitMQ 构建异步任务处理系统

pip install django celery

python
# settings.py
# ...
INSTALLED_APPS = [
    # ...
    'django_celery_results',  # 添加 celery_results 应用
]
CELERY_BROKER_URL = 'amqp://guest:guest@localhost:5672/'
CELERY_RESULT_BACKEND = 'django-db'

python
# celery.py
from __future__ import absolute_import, unicode_literals
import os
from celery import Celery
# 设置默认的 Django 设置模块
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project.settings')
app = Celery('your_project')
# 从 Django 配置中加载 Celery 配置
app.config_from_object('django.conf:settings', namespace='CELERY')
# 自动发现异步任务
app.autodiscover_tasks()

python
# urls.py
from django.urls import include, path
from django_celery_results.views import TaskResultView
urlpatterns = [
    # ...
    path('api/celery/results//', TaskResultView.as_view(), name='task_result'),
    path('api/celery/', include('django_celery_results.urls')),
    # ...
]

python
# tasks.py
from celery import shared_task
from django.core.mail import send_mail
@shared_task
def send_email_task(subject, message, from_email, recipient_list):
    send_mail(subject, message, from_email, recipient_list)

python
from .tasks import send_email_task
def send_email(request):
    subject = 'Hello'
    message = 'This is a test email.'
    from_email = 'admin@example.com'
    recipient_list = ['user@example.com']
    
    send_email_task.delay(subject, message, from_email, recipient_list)
    
    return HttpResponse('Email sent successfully.')