芹菜celery是一个python实现的异步任务队列,可以用于爬虫、web后台查询、计算等等。通过任务队列,当一个任务来临时不再傻傻等待。
他的架构如下:
broker我们的生产者创建任务后会进入celery的任务调度队列中间件broker,broker通过调度规则将消息(任务)调度消息队列,broker依赖第三方队列消息代理如rabbitmq
、redis
等。
广大劳动者,盯着消息队列,当队列中有消息时把它拿过来给处理了。
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
#my_task1.pyfrom .celery import appimport [email protected] 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 [email protected] 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然后返回结果。
#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监控celery1.启动flowercelery -A learn_celery flower
2. 查看web监控 http://ip:5555在Tasks
中可以查看到当前任务队列的状态、参数、接收和启动、执行时间。
在
Dashborad
中查看当前worker节点的相关信息文章有不足的地方欢迎指出。
欢迎收藏、点赞、提问。关注顶级饮水机管理员,除了管烧热水,有时还做点别的。
NEXTcelery的深入了解
celery在django中的使用
总结以上是内存溢出为你收集整理的Python中任务队列-芹菜celery的使用全部内容,希望文章能够帮你解决Python中任务队列-芹菜celery的使用所遇到的程序开发问题。
如果觉得内存溢出网站内容还不错,欢迎将内存溢出网站推荐给程序员好友。
欢迎分享,转载请注明来源:内存溢出
评论列表(0条)