Django调用Celery .delay()后程序冻结无报错，如何修复？

嘿，我之前也碰到过类似的Celery任务阻塞问题，结合你给出的代码和现象，给你几个排查和修复的方向：

1. 先检查Celery Worker的启动命令是否正确 #

很多时候问题就出在Worker的启动参数上，你得确保Worker是对着你的Django项目正确启动的：

• 启动时一定要指定项目的Celery实例，同时打开info级别的日志，这样能看到Worker的详细运行情况，方便定位问题：

  celery -A src worker --loglevel=info
  

  这里的-A src要和你celery.py里app = Celery('src')的名称完全对应，不能错。

2. 确认Redis Broker的连接配置没毛病 #

虽然你说Redis运行正常，但还是得核对Django settings里的Celery配置：

• 确保CELERY_BROKER_URL和CELERY_RESULT_BACKEND（如果用Redis存结果的话）是正确的Redis地址，比如本地默认的话应该是：

  CELERY_BROKER_URL = 'redis://localhost:6379/0'
  CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
  

  如果你的Redis设了密码，要加上认证：redis://:yourpassword@localhost:6379/0，端口和数据库编号也要和你的Redis实例匹配。

3. 删掉tasks.py里多余的Celery初始化代码 #

你的tasks.py里重复导入并初始化了Celery，这会导致Worker和Django项目用的不是同一个Celery实例，任务自然没法正确传递：

• 修改后的tasks.py应该是这样的，只保留shared_task就行：

  import os
  import datetime
  from celery import shared_task
  
  @shared_task
  def testfunc():
      print('EKHASGBDJHGASd')
      return False
  

  因为celery.py已经完成了Celery实例的全局配置，shared_task会自动关联到那个实例，不需要再单独创建Celery对象。

4. 排查任务是否真的被Worker接收 #

你说程序完全冻结，有可能是任务根本没被Worker拿到，或者执行时卡住了。可以给任务加个日志，方便确认：

• 修改testfunc，加上日志输出：

  import logging
  logger = logging.getLogger(__name__)
  
  @shared_task
  def testfunc():
      logger.info("testfunc 任务开始执行")
      print('EKHASGBDJHGASd')
      logger.info("testfunc 任务执行结束")
      return False
  

  然后启动Worker时开info日志，看看有没有这些日志输出，就能判断任务是不是真的被Worker处理了。

5. 确保Django和Celery用的是同一个settings #

你的celery.py里指定了src.settings.local，但启动Worker时可能被其他环境变量覆盖了这个设置。可以先检查一下：

echo $DJANGO_SETTINGS_MODULE

如果输出不是src.settings.local，手动设置后再启动Worker：

export DJANGO_SETTINGS_MODULE=src.settings.local
celery -A src worker --loglevel=info

另外你提到在runserver里会打印一次"2,2"，会不会是你在views里不小心直接调用了testfunc()（同步执行）而不是testfunc.delay()？可以检查下views里的调用逻辑，确保是异步调用。

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