20 python--celery

Celery概述

关于celery的定义，首先来看官方网站：Celery(芹菜) 是一个简单、灵活且可靠的，处理大量消息的分布式系统，并且提供维护这样一个系统的必需工具。

简单来看，是一个基于python开发的分布式异步消息任务队列，持使用任务队列的方式在分布的机器、进程、线程上执行任务调度。通过它可以轻松的实现任务的异步处理， 如果你的业务场景中需要用到异步任务，就可以考虑使用celery， 举几个实例场景中可用的例子:

1. 你想对100台机器执行一条批量命令，可能会花很长时间 ，但你不想让你的程序等着结果返回，而是给你返回 一个任务ID,你过一段时间只需要拿着这个任务id就可以拿到任务执行结果， 在任务执行ing进行时，你可以继续做其它的事情。 　　
2. 你想做一个定时任务，比如每天检测一下你们所有客户的资料，如果发现今天 是客户的生日，就给他发个短信祝福 。　　

Celery 在执行任务时需要通过一个中间人(消息中间件)来接收和发送任务消息，以及存储任务结果，完整的中间人列表请查阅官方网站

PS：任务队列是一种在线程或机器间分发任务的机制。

PS：消息队列的输入是工作的一个单元，称为任务，独立的工作（Worker）进程持续监视队列中是否有需要处理的新任务。 

Celery简介

Celery 系统可包含多个职程和中间人，以此获得高可用性和横向扩展能力，其基本架构由三部分组成，消息中间件（message broker），任务执行单元（worker）和任务执行结果存储（task result store）组成。

1. 消息中间件，Celery本身不提供消息服务，但是可以方便的和第三方提供的消息中间件集成，一般使用rabbitMQ or Redis，当然其他的还有MySQL以及Mongodb。
2. 任务执行单元，Worker是Celery提供的任务执行的单元，worker并发的运行在分布式的系统节点中。
3. 任务结果存储，Task result store用来存储Worker执行的任务的结果，Celery支持以不同方式存储任务的结果，包括Redis，MongoDB，Django ORM，AMQP等。

Celery的主要特点：

1. 简单：一单熟悉了celery的工作流程后，配置和使用还是比较简单的
2. 高可用：当任务执行失败或执行过程中发生连接中断，celery 会自动尝试重新执行任务
3. 快速：一个单进程的celery每分钟可处理上百万个任务
4. 灵活： 几乎celery的各个组件都可以被扩展及自定制

Celery基本工作流程图

1、用户应用程序讲任务通过celery放入Broker中。

　　2、多个worker通过Broker获取任务并执行。

　　3、worker执行完成后，会把任务的结果、状态等信息返回到Broker中存储，供用户程序读取。

PS：Celery 用消息通信，通常使用中间人（Broker）在客户端和职程间斡旋。这个过程从客户端向队列添加消息开始，之后中间人把消息派送给职程。

Celery模块的基本使用

# 利用pip3命令安装celery模块

pip3 install celery

 

用来定义任务列表，这里任务文件的名称叫做task.py(注意后面会用到文件名)。

from celery import Celery
 
app = Celery('task',　　# 是当前模块的名称，这个参数是必须的，这样的话名称可以自动生成
    broker="redis://192.168.100.83:6379/0",     # 中间人的地址
    backend="redis://192.168.100.83:6379/1"　　　# 结果数据存放地址
)
 
@app.task　　　　　# 使用celery标识一个任务，多个任务都需要使用该装饰器　
def add(x,y):　　　　
    return x+y

其他中间件

1 # rabbitmq
2 broker = 'amqp://user:password@ip:5672//'
3 
4 # redis
5 broker = 'redis://passwordf@ip:6379/db'

运行一个worker

elery -A task worker --loglevel=debug
 
# -A参数表示的是Celery APP的名称，task就是APP的名称(应用文件名)
# worker表示是一个执行任务角色
# loglevel=info记录日志类型默认是info

生产者

form ling import add
re = add.delay(10, 20)
print(re.result)    #获取结果
print(re.ready)        #是否处理
print(re.get(timeout=1))        #获取结果
print(re.status)        #是否处理