我这里用的是celery redis的组合,rabbitmq有些重型。

启动的时候报错，查了下文档，问题确认是全局变量一个参数的问题。

[root@devops-ruifengyun ~ ][root@devops-ruifengyun ~ ] celery -A tasks worker --loglevel=debug
Running a worker with superuser privileges when the
worker accepts messages serialized with pickle is a very bad idea!

If you really want to continue then you have to set the C_FORCE_ROOT
environment variable (but please think about this before you do).

User information: uid=0 euid=0 gid=0 egid=0

[root@devops-ruifengyun ~ ]export C_FORCE_ROOT="true"
[root@devops-ruifengyun ~ ] celery -A tasks worker --loglevel=debug
/usr/local/lib/python2.7/dist-packages/celery/platforms.py:762: RuntimeWarning: You are running the worker with superuser privileges, which is
absolutely not recommended!

Please specify a different user using the -u option.

解决办法：

export C_FORCE_ROOT="true"

    293 
–> 294         meta = self._get_task_meta_for(task_id)
    295         if cache and meta.get(‘status’) == states.SUCCESS:
    296             self._cache[task_id] = meta

AttributeError: ‘DisabledBackend’ object has no attribute ‘_get_task_meta_for’

pip install celery-with-redis

测试的代码：

[root@devops-ruifengyun ~ ]cat celeryconfig.py
BROKER_URL = "redis://127.0.0.1:6379/0"
CELERY_IMPORTS = ("tasks.add", )
[root@devops-ruifengyun ~ ] 
[root@devops-ruifengyun ~ ]$ cat test.py 
from tasks import add
if __name__ == '__main__':
    for i in range(100):
        for j in range(100):
            kk=add.delay(i, j)
            kk.ready()
            kk.get()

The full contents of the message body was:
{‘utc’: True, ‘chord’: None, ‘args’: (0, 0), ‘retries’: 0, ‘expires’: None, ‘task’: ‘tasks.add’, ‘callbacks’: None, ‘errbacks’: None, ‘timelimit’: (None, None), ‘taskset’: None, ‘kwargs’: {}, ‘eta’: None, ‘id’: ’48c13a71-ecda-4ef8-98ce-f223a28c0b97′} (209b)
Traceback (most recent call last):
  File “/usr/local/lib/python2.7/dist-packages/celery/worker/consumer.py”, line 455, in on_task_received
    strategies[name](message, body,
KeyError: ‘tasks.add’
[2014-04-23 09:57:16,673: ERROR/MainProcess] Received unregistered task of type ‘tasks.add’.
The message has been ignored and discarded.

Did you remember to import the module containing this task?
Or maybe you are using relative imports?
Please see http://bit.ly/gLye1c for more information.

The full contents of the message body was:
{‘utc’: True, ‘chord’: None, ‘args’: (0, 0), ‘retries’: 0, ‘expires’: None, ‘task’: ‘tasks.add’, ‘callbacks’: None, ‘errbacks’: None, ‘timelimit’: (None, None), ‘taskset’: None, ‘kwargs’: {}, ‘eta’: None, ‘id’: ‘eb1b8772-ca7f-4966-8033-7b4db2a82aa6’} (209b)
Traceback (most recent call last):
  File “/usr/local/lib/python2.7/dist-packages/celery/worker/consumer.py”, line 455, in on_task_received
    strategies[name](message, body,
KeyError: ‘tasks.add’

解决的办法：

最好是用配置文件的方式：

celery --config=celeryconfig --loglevel=INFO

BROKER_URL = 'amqp://'
CELERY_RESULT_BACKEND = 'amqp://'

CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'America/Los_Angeles'
CELERY_ENABLE_UTC = True

CELERY_IMPORTS = ("tasks",)

注释下: 

文章的原文,blog.xiaorui.cc