基于celery + RabbitMQ的异步队列任务

2019年7月23日编辑 3 阅读量:1103

首先，RabbitMQ是个消息队列的实现。
    将消息队列应用到何种场景，得看自己需求。你可以拿它自己实现任务分发（自己实现一个Celery），也可以拿它实现消息推送，甚至只是用它解耦应用的物理结构，等等等等...

其次，Celery是一个任务分发系统。
    任务分发系统目的很明确，就是利用后端待命的无数worker实现一系列任务的快速处理。它跟消息队列的关系不过是利用其在分发者和执行者之间进行任务的发布/订阅。

所以RabbitMQ和redis等具有发布/订阅能力的工具，理论上都能为其所用。
    现在的Celery早已增加了一些对Redis,MongoDB之类的支持。原因是RabbitMQ尽管足够强大，但对于一些相对简单的业务环境来说可能太多（复杂）了一些。这样用户可以有多一些的选择。

celery的介绍 : 
    Celery是一个异步任务队列/基于分布式消息传递的作业队列。它侧重于实时操作，但对调度支持也很好。
celery是用Python编写的，但该协议可以在任何语言实现。它也可以与其他语言通过webhooks实现。
建议的消息代理RabbitMQ的，但提供有限支持Redis, Beanstalk, MongoDB, CouchDB, ，和数据库（使用SQLAlchemy的或Django的 ORM） 。

celery是易于集成Django, Pylons and Flask，使用 django-celery, celery-pylons and Flask-Celery 附加包即可。 
    目前，在Django中使用celery有两种方式：一：使用celery + Django；二：使用celery + Django + django-celery 。这里主要介绍第一种方式(个人感觉会更稳定)。

一、安装RabbitMQ
    1.添加以下地址到/etc/apt/sources.list

```
deb http://www.rabbitmq.com/debian/ testing main
```

2.添加public key：在命令行窗口中输入：

```
$ wget http://www.rabbitmq.com/rabbitmq-signing-key-public.asc sudo apt-key add 
$ rabbitmq-signing-key-public.asc
$ sudo apt-get update
```

3.安装

```
$ sudo apt-get install rabbitmq-server
```

在rabbitmq安装的过程中，需要用到erlang，使用以上方法安装时，都可以自动安装这些必要的包 。常用命令：

```
 #启动
sudo rabbitmq-server start
#关闭
 sudo rabbitmq-server stop
#查看状态
sudo rabbitmqctl status 
```

二、安装celery

```
sudo pip install celery
```

三、配置Django项目
    1、在项目的settings.py中添加配置

```python
# 表面使用rabbitmq做broker， backend
BROKER_URL = 'amqp://guest:guest@localhost:5672/'
CELERY_RESULT_BACKEND = 'amqp'
```

2、在与settings.py同级的目录下新建celery.py

```python
from __future__ import absolute_import
import os
from celery import Celery
from django.conf import settings

# set the default Django settings module for the 'celery' program.  
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'blog_project.settings')  
app = Celery('blog_project')

# Using a string here means the worker will not have to  
# pickle the object when using Windows.  
app.config_from_object('django.conf:settings', namespace='CELERY')  
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

@app.task(bind=True)  
def debug_task(self):  
    print('Request: {0!r}'.format(self.request))  
```

3、还是在同样的目录下配置__init__.py，用来保证django项目运行时能第一时间载入celery

```python
from __future__ import absolute_import

# This will make sure the app is always imported when
# Django starts so that shared_task will use this app.

from .celery import app as celery_app
```

4、使用
    使用时，可以在各个app下，与models.py同一目录下新建tasks.py

```python
from __future__ import absolute_import, unicode_literals
from celery import shared_task

@shared_task
def add(x, y):
    return x + y
```

接着就可以在views.py或者python manage shell中调用，然后在开另一个celery进程接受任务，过程如下：
A、在Django目录（与manage.py同一个目录）下开启celery worker进程

```
celery -A blog_project worker --loglevel=info
```

B、开启Django的shell， python manage shell

```
from blog.tasks import add
r = add.delay(3, 4)
```

这时，可以在celery中看出任务已经提交，并且处理。

但是，以上目录结构仅适用于python manage runserver 8080（或者apache2 + mod_wsgi）这两方式运行，不适用于nginx + uwsgi + wsgi方式。因为后面一种方式中，wsgi与celery.py在同一目录下，会造成celery.py中的from celery import Celery导入的是自身(可以通过import celery; print celery.__path__查看），及时是使用了from __future__ import absolute_import 也会有该错误。而前面的方式中，运行目录不是在wsgi中，所以不会有该错误。所以，如果要在后一种方式中使用，可以将wsgi.py提到外面一层目录中，这时wsgi.py里面的
PROJECT_DIR = dirname(dirname(abspath(__file__)))要去掉一层dirname，然后uwsgi的配置文件bloguwsgi.ini要修改wsgi的目录。

具体代码可以看项目：

```
https://github.com/caitinggui/Blog/commit/4715e3ff388aa61b8852273df185077ce2f92b83
```

warnings.warn('Using settings.DEBUG leads to a memory leak, never '：有这个报错是因为celery版本较低，会默认使用debug模式的原因，升级到4.0.1以上就没有这个问题了。当置顶loglevel=error时，不会有提示。

```
celery -A blog_project worker --loglevel=error
```

ps:RabbitMQ可以实现优先级队列。让高优先级的队列优先被分发。

参考连接： http://python.jobbole.com/81953/
               http://blog.liuts.com/post/248/
               http://charlee.li/django-celery-rabbitmq-intro.html
               http://www.axiaoxin.com/article/71/

基于celery + RabbitMQ的异步队列任务

附件: