在Django框架外运行Celery任务遭遇阻塞问题求助

直接调用testtask(10)正常，但testtask.delay(10)无限阻塞，核心原因是你的脚本只初始化了Django，却没正确初始化Celery实例，导致异步任务无法连接到Broker。

解决步骤： #

1. 导入并初始化项目的Celery实例
   你的Django项目里应该有一个Celery配置文件（通常是raxakprotect/celery.py），里面定义了Celery app实例，包含Broker连接配置。在脚本里必须导入这个实例，确保Celery能读取到Broker信息。

2. 修正交互环境操作流程
   调整后的操作代码如下：

   >>> import os
   >>> os.environ.setdefault("DJANGO_SETTINGS_MODULE", "raxakprotect.settings.dev")        
   >>> import django                                                                 
   >>> django.setup()
   >>> # 导入项目的Celery实例，路径根据你的实际项目结构调整
   >>> from raxakprotect.celery import app
   >>> from test import testtask 
   >>> testtask.delay(10)  # 此时任务应该能正常发送到Worker队列
   

3. 额外检查项

   • 确认settings.py里的CELERY_BROKER_URL配置正确，且对应的Broker服务（Redis/RabbitMQ等）正在运行
   • 确保Celery Worker处于启动状态，且监听的队列和任务默认队列一致
   • 如果任务指定了自定义队列，需要用testtask.apply_async(args=(10,), queue='your_queue_name')明确指定队列

补充说明： #

直接调用testtask(10)是同步执行，不需要和Broker交互，所以能正常运行；而delay()方法需要将任务消息发送到Broker，必须依赖已初始化并配置好的Celery实例，否则会因无法建立Broker连接而无限阻塞。

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