Celery启动Worker报错AttributeError: async 技术求助

我之前在Python 2.7环境下用Celery 4.x时也碰到过一模一样的问题，结合你描述的场景（Mac本地正常、ODROID-C2的Kali环境报错），大概率是依赖版本不兼容或者Python 2.7的兼容性限制导致的，试试下面这几个解决方案：

1. 匹配Celery与Kombu的兼容版本 #

Celery 4.1.0需要特定版本的Kombu才能在Python 2.7下稳定运行，你当前环境的Kombu版本可能过高。执行以下命令安装兼容版本：

pip install kombu==4.1.0

安装完成后重新启动Worker：

celery -A tasks worker --loglevel=info

2. 强制禁用事件循环检测 #

错误栈里的should_use_eventloop方法触发了属性错误，说明Celery在判断是否启用事件循环时出了问题。你可以直接在启动Worker时禁用相关功能：

celery -A tasks worker --loglevel=info --without-gossip --without-mingle --without-heartbeat

或者在你的tasks代码里添加配置，彻底禁用事件循环：

#!/usr/bin/env python
from celery import Celery
from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)

app = Celery('tasks', 
             broker='pyamqp://appuser:xx@c2/appvhost', 
             backend='db+mysql://appuser:xx@c2/pigpen')

# 添加这行配置禁用事件循环
app.conf.worker_disable_event_loop = True

@app.task(bind=True)
def dump_context(self, x, y):
    print('Executing task id {0.id}, args: {0.args!r} kwargs {0.kwargs!r}'.format(self.request))

@app.task
def add(x, y):
    logger.info('Adding {0} + {1}'.format(x, y))
    return x + y

3. 更换RabbitMQ传输协议 #

虽然你的Pika脚本能正常连接RabbitMQ，但Celery默认使用的pyamqp可能在你的设备上存在兼容性问题。试试把broker的协议改成amqp：

app = Celery('tasks', 
             broker='amqp://appuser:xx@c2/appvhost', 
             backend='db+mysql://appuser:xx@c2/pigpen')

4. 清理冲突依赖 #

如果上面的方法都无效，建议彻底卸载现有Celery及相关依赖，重新安装指定版本：

pip uninstall celery kombu pyamqp amqp -y
pip install celery==4.1.0 kombu==4.1.0

问题根源说明 #

这个错误本质是Python 2.7与Celery 4.x的兼容性矛盾：Celery 4.x后期版本逐渐放弃了对Python 2.7的支持，而Kombu（Celery的消息队列依赖）的新版本中使用了async相关属性，在Python 2.7环境下触发了属性错误。你的Mac环境刚好安装了兼容的依赖版本，所以没有出现问题。

附你的问题代码： #

#!/usr/bin/env python
from celery import Celery
from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)

app = Celery('tasks', \
broker='pyamqp://appuser:xx@c2/appvhost', \
backend='db+mysql://appuser:xx@c2/pigpen')

@app.task(bind=True)
def dump_context(self, x, y):
    print('Executing task id {0.id}, args: {0.args!r} kwargs {0.kwargs!r}'.format(self.request))

@app.task
def add(x, y):
    logger.info('Adding {0} + {1}'.format(x, y))
    return x + y

错误栈信息： #

Traceback (most recent call last):
  File "/usr/local/bin/celery", line 11, in <module>
    sys.exit(main())
  File "/usr/local/lib/python2.7/dist-packages/celery/__main__.py", line 14, in main
    _main()
  File "/usr/local/lib/python2.7/dist-packages/celery/bin/celery.py", line 326, in main
    cmd.execute_from_commandline(argv)
  File "/usr/local/lib/python2.7/dist-packages/celery/bin/celery.py", line 488, in execute_from_commandline
    super(CeleryCommand, self).execute_from_commandline(argv)))
  File "/usr/local/lib/python2.7/dist-packages/celery/bin/base.py", line 281, in execute_from_commandline
    return self.handle_argv(self.prog_name, argv[1:])
  File "/usr/local/lib/python2.7/dist-packages/celery/bin/celery.py", line 480, in handle_argv
    return self.execute(command, argv)
  File "/usr/local/lib/python2.7/dist-packages/celery/bin/celery.py", line 412, in execute
    ).run_from_argv(self.prog_name, argv[1:], command=argv[0])
  File "/usr/local/lib/python2.7/dist-packages/celery/bin/worker.py", line 221, in run_from_argv
    return self(*args, **options)
  File "/usr/local/lib/python2.7/dist-packages/celery/bin/base.py", line 244, in __call__
    ret = self.run(*args, **kwargs)
  File "/usr/local/lib/python2.7/dist-packages/celery/bin/worker.py", line 255, in run
    **kwargs)
  File "/usr/local/lib/python2.7/dist-packages/celery/worker/worker.py", line 99, in __init__
    self.setup_instance(**self.prepare_args(**kwargs))
  File "/usr/local/lib/python2.7/dist-packages/celery/worker/worker.py", line 122, in setup_instance
    self.should_use_eventloop() if use_eventloop is None
  File "/usr/local/lib/python2.7/dist-packages/celery/worker/worker.py", line 241, in should_use_eventloop
    self._conninfo.transport.implements.async and
  File "/home/autossh/.local/lib/python2.7/site-packages/kombu/transport/base.py", line 125, in __getattr__
    raise AttributeError(key)
AttributeError: async

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