Celery学习---Celery 分布式队列介绍及安装

Posted 2021-01-02 ftl1012

 Celery介绍和基本使用

Celery 是一个 基于python开发的分布式异步消息任务队列，通过它可以轻松的实现任务的异步处理， 如果你的业务场景中需要用到异步任务，就可以考虑使用celery， 举几个实例场景中可用的例子:

1. 你想对100台机器执行一条批量命令，可能会花很长时间 ，但你不想让你的程序等着结果返回，而是给你返回 一个任务ID,你过一段时间只需要拿着这个任务id就可以拿到任务执行结果， 在任务执行ing进行时，你可以继续做其它的事情。

2. 你想做一个定时任务，比如每天检测一下你们所有客户的资料，如果发现今天 是客户的生日，就给他发个短信祝福

【简单讲：异步  +  定时】

Celery有以下优点

1. 简单：一单熟悉了celery的工作流程后，配置和使用还是比较简单的

2. 高可用：当任务执行失败或执行过程中发生连接中断，celery 会自动尝试重新执行任务

3. 快速：一个单进程的celery每分钟可处理上百万个任务

4. 灵活： 几乎celery的各个组件都可以被扩展及自定制

Celery 在执行任务时需要通过一个消息中间件来接收和发送任务消息，以及存储任务结果， 一般使用rabbitMQ or Redis

Celery是一个上层任务，当有用户请求到来的时候，发送一个任务给celery,这个任务会被中间件Redis/RabbitMQ接收发送给Celery的节点去执行任务，有一个好处就是可以横向的扩展机器去执行任务。

Celery安装使用

Celery安装使用

Celery的默认broker[可理解为中间件Redis/RabbitMQ]是RabbitMQ, 仅需配置一行就可以

	broker_url = ‘amqp://guest:[email protected]:5672//‘

rabbitMQ 没装的话请装一下，安装看这里  http://docs.celeryproject.org/en/latest/getting-started/brokers/rabbitmq.html#id3

使用Redis做broker也可以

安装redis组件

pip install -U "celery[redis]"

配置

Configuration is easy, just configure the location of your Redis database:
app.conf.broker_url = ‘redis://localhost:6379/0‘
Where the URL is in the format of:【如果中间件有认证操作】
redis://:[email protected]:port/db_number
all fields after the scheme are optional, and will default to localhost on port 6379, using database 0.   如果想获取每个任务的执行结果，还需要配置一下把任务结果存在哪
If you also want to store the state and return values of tasks in Redis, you should configure these settings:
app.conf.result_backend = ‘redis://localhost:6379/0‘

Celery的基本操作+异步使用　　

Win7下安装celery模块

pip3 install celery

[测试发现celery安装完成后Win7的cmd可以用celery，不需要添加]
如果需要手动添加，则找到celery的安装路径，写入Win7的path里即可。
E:PyCharm 2017.2.4Python3.2.5Libsite-packagesceleryin

 

Ubuntu下安装

首先安装pip3

sudo apt-get install python3-pip  

安装Celery

pip3 install -i https://pypi.tuna.tsinghua.edu.cn/simple celery

  Ubuntu下安装成功

Ubuntu下运行报错，缺少Redis

运行Celery任务后报错

ImportError: Missing redis library (pip install redis)

安装Redis连接模块：

pip3 install redis   

启动Celery Worker来开始监听并执行任务

celery -A  Celery的Py文件名 worker --loglevel=info

后台Celery任务的运行

操作前期条件： 安装并启动Redis

myCelery.py

from celery import Celery
# 定义了一个Celery的App
app = Celery(‘tasks‘,
             broker=‘redis://192.168.2.105‘,   # Celery和用户请求的中间代理
             backend=‘redis://192.168.2.105‘)  # 接收Celery返回结果的

@app.task       # 函数变成一个Celery的任务，调用celery实现异步任务
def add(x, y):
    print("running...", x, y)
    return x + y

运行：

celery -A myCelery worker --loglevel=info 【写文件名称即可】

运行结果[Win]

运行结果[Linux]

注： celery文件运行起来后只能接收和执行任务[等待任务状态...]，还需要用户发送任务

前台Celery任务的发布

Linux下发布任务

进入文件所在的路径下

[email protected]:~/Celery$ cd /home/omc/Celery

进入Python环境

[email protected]:~/Celery$ python

在Python环境下导入文件的函数

Python 3.5.2 (default, Nov 23 2017, 16:37:01) 
[GCC 5.4.0 20160609] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> from myCelery import add
>>> add.delay(100,100)
<AsyncResult: 22631adf-e54f-4718-99a1-1e76b5c1575f>
>>> add(100,300)      
running... 100 300
400

注：中间件Redis只负责存储了中间的消息

打开Redis查看存储的消息内容

[email protected]:~$ redis-cli

127.0.0.1:6379> get celery-task-meta-ebad12b8-d4ff-461b-b6e5-0dab4eaba855

>>> r = add.delay(20,20000)
>>> r.

>>> r.get()
20020

返回对象的其他方法的使用：

>>> r = cmd_Celery.my_cmd.delay("df -h")
The ready() method returns whether the task has finished processing or not:
>>> r.ready()
False

You can wait for the r to complete, but this is rarely used since it turns the asynchronous call into a synchronous one:
>>> r.get(timeout=1)  # 设置超时时间
8

In case the task raised an exception, get() will re-raise the exception, but you can override this by specifying the propagate argument:
>>> r.get(propagate=False)   #  propagate 扩展，添加propagate后会对报错进行格式化输出

If the task raised an exception you can also gain access to the original traceback:
>>> r.traceback     # 报错调试用
…

当Celery有多个worker的时候[默认是随机分配的]：

前台发布：

后台有2个Celery的worker

  work1:

Work2: 

Celery异步执行命令

操作前期条件： 安装并启动Redis

cmd_Celery.py

from celery import Celery
# 定义了一个Celery的App
app = Celery(‘tasks‘,
             broker=‘redis://192.168.2.105‘,
             # redis://:[email protected]:port/db_number  有密码认证的连接
             # broker=‘redis://:密码@192.168.2.105:6379/0‘,
             backend=‘redis://192.168.2.105‘)  # 接收Celery返回结果的

# 函数变成一个Celery的任务，调用celery实现异步任务
import subprocess
import time
@app.task
def my_cmd(cmd):
    print(‘Celery of CMD:‘, cmd)
    time.sleep(5)   # 判断是否是异步的标志，会卡5秒后执行cmd的命令
   cmd_obj = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    return cmd_obj.stdout.read().decode("utf-8")  # 返回结果必须是可JSON的，需要编码

后台启动Celery任务：

[email protected]:~/Celery$ celery -A cmd_Celery worker --loglevel=debug 

前台客户端发送命令：

>>> python3
>>> import cmd_Celery
>>> r = cmd_Celery.my_cmd.delay("df -h")
>>> r.get()

后台Celery服务器端执行任务：

如果返回结果不是可JSON会报错：

默认返回的是byte类型的，需要进行解码

后台Celery的终止

2次Ctrl+C

后台启动/停止多个Celery的worker

前台启动命令： celery -A 项目名worker -loglevel=info 

后台启动命令： celery multi start w1 -A 项目名 -l info 

后台重启命令： celery multi start w1 -A 项目名 -l info 

后台停止命令： celery multi stop w1 -A 项目名 -l info 

前后台的区别： 后台是mult启动