Celery 初步使用心得

一、 基本介绍

Celery是一个专注于实时处理和任务调度的分布式任务队列。所谓任务就是消息，消息中的有效载荷中包含要执行任务需要的全部数据。

使用Celery常见场景：

1. Web应用。当用户触发的一个操作需要较长时间才能执行完成时，可以把它作为任务交给Celery去异步执行，执行完再返回给用户。这段时间用户不需要等待，提高了网站的整体吞吐量和响应时间。
2. 定时任务。生产环境经常会跑一些定时任务。假如有上千台的服务器、上千种任务，定时任务的管理会很困难，Celery可以帮助我们快速在不同的机器设定不同种任务。
3. 其他可以异步执行的任务。为了充分提高网站性能，对于请求和响应之外的那些不要求必须同步完成的附加工作都可以异步完成。比如发送邮件/短信、推送消息、清理/设置缓存等。

Celery特性：

• 方便地查看定时任务的执行情况，比如执行是否成功、当前状态、执行任务花费的时间等。
• 可以使用功能齐备的管理后台或者命令行添加、更新、删除任务。
• 方便把任务和配置管理相关联。
• 可选多进程、Evenlent和Gevent三种模式并发执行。
• 提供多种错误处理机制。
• 提供多种任务原语，方便实现任务分组、拆分和调用链。
• 支持多种消息代理和存储后端。

Celery架构图：

产生任务的方式有两种：

1. 发布者发布任务（Web应用）
2. 任务调度按期发布任务（定时任务）

Celery组件介绍：

• Celery Beat： 任务调度器，Beat进程会读取配置文件的内容，周期性地将配置中到期需要执行的任务发送给任务队列。
• Celery Worker： 执行任务的消费者，通常会在多台服务器运行多个消费者来提高执行效率。
• Broker： 消息代理，或者叫做消息中间件，接受任务生产者发送过来的任务消息，存进队列在按序分发给任务消费方。
• Producer： 调用Celery提供的API、函数或者装饰器而产生任务并交给任务队列处理的都是任务生产者。
• Result Backend： 任务处理完后保存状态信息和结果，以供查询。Celery默认支持Redis、RabbitMQ、MongoDB、Django ORM SQLAlchemy等方式。

Celery目前支持RabbitMQ、Redis、MongoDB、Beanstalk、Zookeeper、SQLAlchemy等作为消息代理，但用于生产环境只有RabbitMQ和Redis，官方推荐RabbitMQ来作为Celery的消息代理。

在客户端和消费者之间传输数据需要序列化和反序列化，Celery支持的序列化方案有pickle，json，yaml，msgpack，一般使用json

 二、安装配置Celery

为了提供更高的性能，采用如下方案：

• 选择RabbitMQ作为消息代理。
• RabbitMQ的Python客户端选择librabbitmq这个C库。
• 选择Msgpack做序列化
• 选择Redis做结果存储

sudo apt-get install rabbitmq-server
sudo apt-get install redis-server
sudo pip install "celery[librabbitmq,redis,msgpack]"

三、示例演示

项目目录结构

tree project
project
├── celeryconfig.py
├── celery.py
├── __init__.py
└── tasks.py

主程序celery.py:

#!/usr/bin/env python
# -*- coding:utf-8 -*-

# 拒绝隐式引入，因为celery.py的名字和celery的包名冲突，需要使用这条语句让程序正确运行
from __future__ import absolute_import

from celery import Celery

# app是 Celery类的实例，创建的时候添加了project.tasks这个模块，也就是包含了project/tasks.py这个文件
app = Celery('project', include=['project.tasks'])

# 把Celery配置存放进project/celeryconfig文件，使用app.config_from_object加载配置
app.config_from_object('project.celeryconfig')


if __name__ == '__main__':
    app.start()

任务函数文件tasks.py:

#!/usr/bin/env python
# -*- coding:utf-8 -*-


from __future__ import absolute_import

from project.celery import app

# 让任务函数生效的方法是添加app.task装饰器
@app.task
def add(x, y):
    return x + y

配置文件celeryconfig.py:

# -*- coding:utf-8 -*-
BROKER_URL = 'amqp://guest:guest@localhost:5672//'  # 使用RabbitMQ作为消息代理
CELERY_TASK_PROTOCOL = 1  # 现在celery升级到了4.0，是老版本的librabbitmq与最新的celery4.0 Message Protocol协议不兼容，celery4.0默认使用Task messages Version 2 ，而librabbitmq使用Task messages Version 1
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0' # 把结果存在Redis
CELERY_TASK_SERIALIZER = 'msgpack'  # 任务序列化肯反序列化使用msgpack方案
CELERY_RESULT_SERIALIZER = 'json'   # 读取任务结果一般性能要求不高，所以使用可读性更好的json
CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24  # 任务过期时间
CELERY_ACCEPT_CONTENT = ['json', 'msgpack'] # 指定接收的内容类型

启动消费者：

celery -A project.celery  worker -l info

终端界面提供了消息代理和存储结果的地址、并发数量、任务列表、交换类型等

开启另一个终端，用ipython调用add函数

In [1]: from project.tasks import add

In [2]: r = add.delay(1,3)

In [3]: r
Out[3]: <AsyncResult: a14d2045-ad40-4240-bbcf-1a8f07899485>

In [4]: r.result
Out[4]: 4

In [5]: r.status
Out[5]: u'SUCCESS'

In [6]: r.successful()
Out[6]: True

In [7]: r.backend
Out[7]: <celery.backends.redis.RedisBackend at 0x7faae433a450>  # 保存在redis中

任务的task_id根据上面提到的task_id获得，可以用下面方法获得结果

# 方法一：

In [9]: task_id = 'a14d2045-ad40-4240-bbcf-1a8f07899485'

In [10]: add.AsyncResult(task_id).get()
Out[10]: 4

# 方法二：

In [12]: from celery.result import AsyncResult

In [13]: AsyncResult(task_id).get()
Out[13]: 4 

四、Django中测试Celery异步邮件

下面在django中模拟一下如何用celery：

首先先看没有用celery的程序

        

这就造成不良好的用户体验了，那你先给用户返回程序执行成功，再在后台执行这5秒

下面就用selery

1.安装selery ，不写版本号默认最新，新版本只需要安装一个celery

pip install celery==3.1.25

pip install celery-with-redis==3.0

pip install django-celery==3.2.1

2.在应用目录下创建名为task.py的文件，该文件用于封装耗时任务的

3.配置setting.py

  3.1注册进app

  INSTALLED_APP=[

    . . . .  +

 ‘djcelery’]

   3.2 再加上，

   import djcelery

   djcelery.setup_loader()

   BROKER_URL='redis://:密码@数据库ip地址:6379/0'      #格式不能错 ，注意分号

   CELERY_IMPORTS=（‘myApp.task’）        #任务文件

4.迁移，生成celery需要的表

python manage.migrate

5.将名为celery.py的文件加入到同工程目录同名的目录下，这个文件是官网给我们写好的

from __future__ import absolute_import

import os
from celery import Celery
from django.conf import settings

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'whthas_home.settings')

app = Celery('project')     #这里改成自己工程名
app.config_from_object('django.conf:settings')
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)


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

6.修改与工程目录同名目录下的__init__.py文件，上图红色箭头指处

from project.celery import app as celery_app

7.将耗时程序封装成任务，在task.py文件里

在view.py文件里

这时你去跑程序，会发现很快，但却没有打印开始与结束这两句，其实它没有执行到耗时程序，还缺最后一步

8.启动redis，确保服务器启动

9.启动worker，在黑窗口进入工程目录下，执行命令

python manage.py celery worker --loglevel=info

访问浏览器，就可以看到很快的返回程序执行成功，而耗时程序在黑窗口进行打印：开始程序执行，等待5秒，结束程序执行

相互不影响

浏览器返回

 黑窗口打印