爱站程序员基地一、关于celery
芹菜celery是一个python实现的异步任务队列，可以用于爬虫、web后台查询、计算等等。通过任务队列，当一个任务来临时不再傻傻等待。
他的架构如下：
- Broker
我们的生产者创建任务后会进入celery的任务调度队列中间件Broker，Broker通过调度规则将消息（任务）调度消息队列，Broker依赖第三方队列消息代理如
rabbitmq
、
redis
等。
- Worker
广大劳动者，盯着消息队列，当队列中有消息时把它拿过来给处理了。
- Backend
用于结果存储经worker处理的结果，比如常用的数据库等。
#### 使用celery
在本文中咱们使用
rabbitmq
（celery推荐）作为消息代理中间件。
我们创建的celery目录如下
learn_celery/...celery_env/...celery.py...my_task1.py...my_task2.py...task1_run.py...task2_run.py
1. 创建虚拟环境并安装celery、flower（web监控），这里不做赘述。
2.安装咱们的消息队列中间件
rabbitmq
这里以docker的方式运行并配置，指定主机名为
rabbit
（rabbitmq是以主机名来访问的，所以这是必须的），容器名称为
celery_rabbitmq
docker run -d -p 5672:5672 -h rabbit --name celery_rabbitmq rabbitmq
添加用于celery访问的用户，以及配置
configure
、
write
和
read
权限，在下面我们配置rabbit_user拥有所有配置、写入和读取权限。
docker exec -it celery_rabbitmq rabbitmqctl add_user rabbit_user rabbit_passdocker exec -it celery_rabbitmq rabbitmqctl add_vhost rabbit_vhostdocker exec -it celery_rabbitmq rabbitmqctl set_user_tags rabbit_user celerydocker exec -it celery_rabbitmq rabbitmqctl set_permissions -p rabbit_vhost rabbit_user ".*" ".*" ".*"
3.创建celery应用
#celery.pyfrom celery import Celerybroker_rabbitmq="amqp://rabbit_user:rabbit_pass@i-k9pwet2d/rabbit_vhost"app=Celery("learn_celery",broker=broker_rabbitmq,backend="rpc://",include=["learn_celery.my_task2","learn_celery.my_task2"])
我们通过创建app来实例化Celery，项目包的名称为
learn_celery
，通过
broker_rabbitmq
来连接rabbitmq，rabbitmq的amqp协议格式为
amqp://userid:password@hostname:port/virtual_host
由于我们是在docker中启动的rabbitmq，所以我们的hostname应该为宿主机的hostname。
指定后端通过rpc回传数据，include加载带worker处理的任务
learn_celery.my_task1
、
learn_celery.my_task2
4.创建两个任务(消息)
#my_task1.pyfrom .celery import appimport time@app.taskdef args_add1(x,y):print("start task no.1 now!")time.sleep(10)print("task no.1 end!")return x+y#my_task12.pyfrom .celery import appimport time@app.taskdef args_add2(x,y):print("start task no.2 now!")time.sleep(20)print("task no.2 end!")return x+y
在这里我们导入了celery中的app，并用它来装饰我们的方法
args_add
，在args_add中模拟任务处理时间分别为10s、20s然后返回结果。
5.发送任务给celery
#tasks1_run.pyfrom .my_task1 import args_add1import timereslut=args_add1.delay(11,22)print("task over?{}".format(reslut.ready()))print("task reslut:{}".format(reslut.result))time.sleep(15)print("task over?{}".format(reslut.ready()))print("task reslut:{}".format(reslut.result))#tasks2_run.pyfrom .my_task2 import args_add2import timereslut=args_add2.delay(33,44)print("task over?{}".format(reslut.ready()))print("task reslut:{}".format(reslut.result))time.sleep(25)print("task over?{}".format(reslut.ready()))print("task reslut:{}".format(reslut.result))
关于任务的
delay
，官方文档（参考）是这样描述的，我把它理解为发送任务给celery或者celery调用待进来的任务。
reslut.ready()
返回任务执行是否执行完成
True
or
False
reslut.result
返回任务执行结果
我们在任务进入celery和结束分别检查一次。
二、看看结果
1.启动worker
进入learn_celery的父目录。启动learn_celery的这个应用worker，并指定并发数为10个
celery -A learn_celery worker --loglevel=info --concurrency=10
若celery连接rabbitmq正常，我们可以看到如下的info
2.执行任务
为了便于观察，我们另外开启一个窗口2，到learn_celery父目录运行task1_run模块
python -m learn_celery.tasks1_run
开启窗口3，到learn_celery父目录运行task2_run模块
python -m learn_celery.tasks2_run
可以看到经过各自任务的等待时间后，两个任务都顺利执行结束，并得到结果，接下来我们到worker上看一下info
由于celery的并发性，收到任务马上被调入执行，任务1耗时10s结果为33，任务2耗时20s结果为77
三、使用Flower监控celery
1.启动flower
celery -A learn_celery flower
2. 查看web监控 http://ip:5555
在
Tasks
中可以查看到当前任务队列的状态、参数、接收和启动、执行时间。
在
Dashborad
中查看当前worker节点的相关信息
文章有不足的地方欢迎指出。
欢迎收藏、点赞、提问。关注顶级饮水机管理员，除了管烧热水，有时还做点别的。
NEXT
-
celery的深入了解
-
celery在django中的使用
