celery 使用详解

celery

是啥？

由python 编写 的异步生产者消费者设计模式下 的实例

举个例子：

现有两个进程 生产者进程A 消费者进程B

现在的情况是

逻辑推导：

A 产出栗子 B 要吃栗子 那么这两个进程必然是 B依赖于A 耦合度很高且是一个耗时操作

 

B -----> （发送请求给A）------->(等待A 产出栗子也许会很久)------->(A响应栗子给B)------->（B得到栗子）

 

B 可能是个很多服务的集成后台之类很忙大忙人不想一直等等等

 

那么 celery 的任务就是 替B 去等

逻辑推导：

A 产出栗子 B 要吃栗子 C celery

B (替我去取栗子)-----> C（发送请求给A）------->(等待A 产出栗子也许会很久)------->(A响应栗子给C)------->（B得到栗子） (C 可以去把栗子存在一个地方B直接去取就好了)

 

那么celery 的本质知道了：一个中间人角色 ，类似快递小哥，跑腿的

 

作用：

1 防止线程阻塞提高性能

2低耦合解耦 高内聚 高复用

好处 ::每分钟可以实现数以百万的任务

特点: 使用消息队列(broker )在客户端和消费者之间协调

主体在消息队列 还有 协调 两端可以有多个

 

 

好的，现在说celery 的结构

Celery的架构由三部分组成，消息中间件（message broker），任务执行单元（worker）和任务执行结果存储（task result store）组成。

消息中间件（message broker）： 这个很好理解 ，B指派给C 的任务可能不只一件，所以celery 要把任务存起来一件一件去执行 ，存储一般是用消息队列结构的，celery 本身是不带存储空间的需要指派位置存 包括，RabbitMQ,Redis,MongoDB等

ps : 队列也是一种数据结构 , 表现形式为一段进入,一段出 先进先出, 不能再中间插入 ,类似管道

 

任务执行单元（worker）： Worker是Celery提供的任务执行的单元，worker并发的运行在分布式的系统节点中 （快递小哥本体）

 

任务执行结果存储（task result store）：小哥把东西要放到快递柜咯 这里也要用到其他服务 包括Redis，MongoDB，Django ORM，AMQP等

 

简单来说就是 broker 收集任务（收件）worker 执行任务 存在 task result store （派送）

 

这里主要使用redis

celery 简单配置

 

# 安装celery
sudo pip install celery -i https://pypi.douban.com/simple
------配置信息  tasks.py-------
BROKER_URL = 'redis://localhost:6379/0'
# 格式 redis://:password@hostname:port/db_number 不过redis 一般不设密码

from celery import Celery
app = Celery("tasks", broker=BROKER_URL)

@app.task
def add(x,y):
    return x,y

----------启动------
$ celery -A tasks worker --loglevel=info


#查询文档，了解到该命令中-A参数表示的是Celery APP的名称，这个实例中指的就是tasks.py,后面的tasks就是APP的名称，worker是一个执行任务角色，
后面的loglevel=info记录日志类型默认是info,这个命令启动了一个worker,用来执行程序中add这个加法任务（task）。

启动成功效果展示

celery 配合 django 使用

项目结构

项目结构

1安装

pip install -U celery

 

2创建Celery 实例加载配置

2.1定义 celery_tasks包

2.2 创建celery 实例 启动文件

main.py
------------
# celery启动文件(独立于项目的启动文件单独启动)
from celery import Celery # 


# 创建celery实例
celery_app = Celery('abc')
# 加载配置
celery_app.config_from_object('celery_tasks.config')

 2.3 加载 celery配置 (celery_tasks.config)

# 指定消息队列的位置redis上存放队列
broker_url = 'redis://192.168.103.210/7'

解耦出来的业务 可以放在celery 的子任务夹里面

注册任务

-----回到main 文件-------
..........
.......
# 自动注册celery任务
celery_app.autodiscover_tasks(['celery_tasks.sms'])

定义任务

案例: 容联云通信接口调用

# 加上装饰器代表载入进去   name：异步任务别名 
@celery_app.task(name='ccp_send_sms_code')
def ccp_send_sms_code(self, mobile, sms_code):
    """
    发送短信异步任务
    :param mobile: 手机号
    :param sms_code: 短信验证码
    :return: 成功0 或 失败-1
    """

    send_ret = CCP().send_template_sms(mobile, [sms_code, constants.SMS_CODE_REDIS_EXPIRES // 60], constants.SEND_SMS_TEMPLATE_ID)

    return send_ret

启动clerery

$ celery -A celery_tasks.main worker -l info

tornado + celery

因为tornado web框架 部分基本仿照了django 所以celery 使用和 django 也是类似的

 结构·

-------config 文件--------------
# celery
BROKER_URL = 'redis://127.0.0.1:6379/2'
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/2'
CELERY_TIMEZONE = 'Asia/Shanghai'
CELERY_TASK_SERIALIZER = 'pickle'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24
CELERY_ACCEPT_CONTENT = ['json', 'pickle']

CELERY_IMPORTS = (
    'celery_tasks.remove_files.tasks',

)

-----main --------
from celery import Celery
from . import config

import os


app = Celery('task')
app.config_from_object(config)

启动clerery

$ celery -A celery_tasks.main worker -l info

现在celery 服务全部启用了 只需要在你的项目里面导包调用他就好了

# 解耦处调用该方法时 delay 是celry 自带的方案可以监听任务的状态
ccp_send_sms_code.delay(**kwars)