Celery 初步使用心得

1、 基本介绍

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

 2、安装配置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]"

3、示例演示

项目目录结构

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 

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秒，结束程序执行

相互不影响

浏览器返回

 

 黑窗口打印