Celery 5.2.7+RabbitMQ 3.8.19集成异常：任务Unacked超时转Ready

问题现象 #

在Django 4.1.5环境中集成Celery 5.2.7与RabbitMQ 3.8.19，提交的任务进入celery队列后显示Unacked状态，30分钟后转为Ready状态，并抛出错误：

amqp.exceptions.PreconditionFailed: (0, 0): (406) PRECONDITION_FAILED - delivery acknowledgement on channel 1 timed out. Timeout value used: 1800000 ms. This timeout value can be configured, see consumers doc guide to learn more

直接调用任务方法（不使用.delay()）时一切正常，无法定位配置错误点。

环境与启动信息 #

启动Celery的命令：

$ celery -A mylocal worker -l info

启动日志：

-------------- celery@DEV2 v5.2.7 (dawn-chorus)
--- ***** -----
-- ******* ---- Windows-1
- *** --- * ---
- ** ---------- [config]
- ** ---------- .> app:         mylocal:0x18205ade830
- ** ---------- .> transport:   amqp://<RMQLogin>:**@<RMQ_HOST>:<RMQ_PORT>/<RMQ_VIRTUAL_HOST>
- ** ---------- .> results:     disabled://
- *** --- * --- .> concurrency: 12 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** -----
-------------- [queues]
                .> celery           exchange=celery(direct) key=celery


[tasks]
  . myapp.tasks.send_spam_email

项目结构 #

mylocal
------|
------mylocal
------------|
------------- __init__.py
------------- celery.py
------------- settings.py
------myapp
----------|
---------- templates
--------------|
---------------myapp
-----------------|
--------------------celery.html
---------- admin.py
---------- forms.py
---------- models.py
---------- service.py
---------- tasks.py
---------- urls.py
---------- views.py

核心代码片段 #

__init__.py #

from .celery import app as celery_app

__all__ = ('celery_app', )

celery.py #

import os
from celery import Celery
from celery.schedules import crontab

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'mylocal.settings')

app = Celery('mylocal')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()    

settings.py（初始配置） #

RMQ_HOST = '10.10.10.10'
RMQ_PORT = '5672'
RMQ_VIRTUAL_HOST = 'VHost'
RMQ_LOGIN = 'RMQLogin'
RMQ_PASSWORD = 'RMQPassword'
CELERY_BROKER_URL = 'amqp://' + RMQ_LOGIN + ':' + RMQ_PASSWORD + '@' + RMQ_HOST + ':' + RMQ_PORT + '/' + RMQ_VIRTUAL_HOST

tasks.py #

from mylocal.celery import app
from .service import send

@app.task
def send_spam_email(user_email):
    send(user_email)

views.py（任务触发逻辑） #

from django.urls import reverse_lazy
from django.views.generic import CreateView

from .forms import ContactForm
from .models import Contact
from .tasks import send_spam_email


class ContactView(CreateView):
    model = Contact
    form_class = ContactForm
    success_url = reverse_lazy('myapp:celery')
    template_name = 'myapp/celery.html'

    def form_valid(self, form):
        form.save()
        send_spam_email.delay(form.instance.email)
        return super().form_valid(form)

已尝试的修改（问题未解决） #

在settings.py中添加：

CELERY_RESULT_BACKEND = 'rpc://'

修改后的启动命令：

$ celery -A mylocal worker -l info -E

修改后的启动日志：

-------------- celery@DEV2 v5.2.7 (dawn-chorus)
--- ***** -----
-- ******* ---- Windows-1
- *** --- * ---
- ** ---------- [config]
- ** ---------- .> app:         mylocal:0x1e64be5e800
- ** ---------- .> transport:   amqp://<RMQLogin>:**@<RMQ_HOST>:<RMQ_PORT>/<RMQ_VIRTUAL_HOST>
- ** ---------- .> results:     rpc://
- *** --- * --- .> concurrency: 12 (prefork)
-- ******* ---- .> task events: ON
--- ***** -----
-------------- [queues]
                .> celery           exchange=celery(direct) key=celery

....
[2023-03-09 12:09:43,316: INFO/MainProcess] Task myapp.tasks.send_spam_email[5cebb2d5-5d19-44e5-b638-f4f86b201df4] received
[2023-03-09 12:09:43,317: INFO/MainProcess] Task myapp.tasks.send_spam_email[f8d7243b-d75e-46c8-b733-4ec080500e60] received
[2023-03-09 12:09:43,318: INFO/MainProcess] Task myapp.tasks.send_spam_email[03ce1b24-4d97-41c5-8765-cd71850d452f] received

解决方案 #

1. 排查任务执行阻塞 #

任务Unacked状态说明worker已接收任务但未发送确认回执，核心原因是任务执行过程中卡住：

• 补全Django邮件配置：本地直接调用send_mail正常，但Celery worker可能未继承系统代理或邮件配置，导致邮件发送阻塞。在settings.py中添加完整SMTP配置：

  EMAIL_BACKEND = 'django.core.mail.backends.smtp.EmailBackend'
  EMAIL_HOST = 'smtp.gmail.com'
  EMAIL_PORT = 587
  EMAIL_USE_TLS = True
  EMAIL_HOST_USER = 'mymail@gmail.com'
  EMAIL_HOST_PASSWORD = '你的Gmail应用密码'  # 需开启两步验证后生成，非登录密码
  

• 给任务添加超时限制：避免任务无限阻塞，在tasks.py中设置：

  @app.task(time_limit=30, soft_time_limit=25)
  def send_spam_email(user_email):
      send(user_email)
  

2. 调整Celery与RabbitMQ的ACK策略 #

• 开启延迟确认+限制预取数：修改settings.py，让worker执行完任务再发送ACK，同时限制预取任务数量，避免多任务堆积阻塞：

  CELERY_ACKS_LATE = True
  CELERY_WORKER_PREFETCH_MULTIPLIER = 1
  

• （可选）调整RabbitMQ超时：若任务确实需要长时间运行，可修改RabbitMQ的rabbitmq.conf配置：

  consumer_timeout = 3600000  # 单位毫秒，设置为1小时
  

3. 解决Windows环境兼容性问题 #

Celery的prefork模式在Windows下存在诸多限制，改用协程模式：

• 安装eventlet：

  pip install eventlet
  

• 用eventlet启动worker：

  celery -A mylocal worker -l info -P eventlet
  

内容的提问来源于stack exchange，提问作者Mikhail