Flask集成Celery无法连接Redis 回退默认AMQP地址报连接被拒

核心问题是Celery实例的初始化逻辑和作用域完全写错了：你在代码里反复创建局部的Celery实例，全局共享的、启动worker时实际加载的那个Celery实例从来没拿到过Redis连接配置，自然会回退到默认的AMQP（RabbitMQ）地址5672端口，最终报连接拒绝。

具体错误点 #

• 顶层无效初始化
  __init__.py最顶部的初始化代码完全无效：

  from app.common.globals import celery
  celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'], backend=app.config['CELERY_BACKEND_URL'])
  celery.conf.update(app.config)
  

  这里导入的app是Flask的current_app代理对象，执行这段代码时没有激活任何Flask应用上下文，current_app根本指向不了真实的Flask实例。更关键的是，你这里是给当前模块的局部变量celery重新赋值，根本没有修改从globals导入的那个全局celery实例。
• 工厂函数内创建的是局部实例
  create_app函数里的Celery初始化同样有问题：

  with app.app_context():
      from app.common.globals import celery
  
  celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'], backend=app.config['CELERY_BACKEND_URL'])
  celery.conf.update(app.config)
  

  你在这里又新建了一个Celery对象，赋值给create_app函数内的局部变量celery，和globals里的全局实例没有任何关系。函数执行完这个局部实例就会被回收，其他地方导入的celery始终是globals里那个初始化时没传任何参数的空Celery()实例——这个空实例默认的broker就是AMQP地址，完全读不到你配置的Redis地址。
• 测试对象搞错了
  你打印配置看到Redis地址的测试代码，打印的是create_app内部新建的局部celery实例，不是worker启动时加载的全局实例，所以才会出现“配置看起来对但连不上”的矛盾现象。

修复步骤 #

不要反复新建Celery实例，所有地方共用globals里定义的全局实例，在工厂函数里给这个全局实例加载配置即可：

1. 精简app/common/globals.py，只保留全局实例定义，删掉没用的current_app导入：

   from celery import Celery
   celery = Celery()
   

2. 删掉__init__.py顶部那段无效的顶层Celery初始化代码，修改create_app逻辑，直接给导入的全局celery实例更新配置，不要新建实例：

   from flask import Flask
   from app.common.globals import celery
   from config import Config
   import logging
   
   def create_app(config_class=Config):
       app = Flask(__name__)
       app.config.from_object(config_class)
       logging.basicConfig(level=logging.INFO)
       if app.config['DEBUG']:
           app.logger.setLevel(logging.INFO)
   
       # 给全局共享的celery实例加载配置
       celery.conf.update(
           broker_url=app.config['CELERY_BROKER_URL'],
           result_backend=app.config['CELERY_BACKEND_URL']
       )
       celery.conf.update(app.config)
   
       # 后续注册蓝图、绑定扩展的逻辑正常写在这里
       
       return app
   

3. 修改启动入口wsgi.py，导出全局celery实例，保证启动worker时能加载到已经完成配置的实例：

   from app import create_app
   from app.common.globals import celery # 导出供celery worker加载
   
   app = create_app()
   debug = app.config['DEBUG']
   
   if __name__ == "__main__":
       app.run(host='127.0.0.1', port=5000, debug=debug, threaded=True)
   

4. 启动Celery worker时指定正确的实例路径，确保先执行create_app完成配置加载：

   celery -A wsgi:celery worker --loglevel=info
   

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