Celery从入门到放弃

1.定义&＃xff1a;

    Celery是一个异步的任务队列&＃xff08;也叫做分布式任务队列&＃xff09;

2.工作结构

    Celery分为3个部分

    &＃xff08;1&＃xff09;worker部分负责任务的处理&＃xff0c;即工作进程&＃xff08;我的理解工作进程就是你写的python代码&＃xff0c;当然还包括python调用系统工具功能&＃xff09;

    &＃xff08;2&＃xff09;broker部分负责任务消息的分发以及任务结果的存储&＃xff0c;这部分任务主要由中间数据存储系统完成&＃xff0c;比如消息队列服务器RabbitMQ、redis、

Amazon SQS、MongoDB、IronMQ等或者关系型数据库&＃xff0c;使用关系型数据库依赖sqlalchemy或者django的ORM

    &＃xff08;3&＃xff09;Celery主类&＃xff0c;进行任务最开始的指派与执行控制&＃xff0c;他可以是单独的python脚本&＃xff0c;也可以和其他程序结合&＃xff0c;应用到django或者flask等web框架里面以及你能想到的任何应用

3.话不多说&＃xff0c;用起来

    &＃xff08;1&＃xff09;安装Celery(要安装celery3版本&＃xff0c;4版本改动较大没测试)

#pip install celery&＃61;&＃61;3.1.17

    &＃xff08;2&＃xff09;broker部分此处使用安装好的redis服务6380端口的db0作为消息队列&＃xff0c;普通redis服务的安装此处不做介绍

  &＃xff08;3&＃xff09;Celery的使用一&＃xff08;单独脚本调用&＃xff0c;简单方便&＃xff09;

        #注&＃xff1a;不考虑任务的结果存储情况

         <1>/tmp/tasks.py&＃xff08;实际脚本中不要写中文注释&＃xff09;

#!/usr/bin/env python

# -*- coding&＃61;utf-8 -*-

from celery import Celery

from celery import platforms

#用于开启root也可以启动celery服务&＃xff0c;默认是不允许root启动celery的

platforms.C_FORCE_ROOT &＃61; True

#创建一个celery实例&＃xff0c;传递进去的第一个参数tasks必须是本文件的文件名tasks&＃xff0c;指定broker为本机redis6380服务

celery &＃61; Celery(&＃39;tasks&＃39;, broker&＃61;&＃39;redis://localhost:6380/0&＃39;)

#使用celery实例的task装饰器装饰add函数&＃xff0c;此处的add函数可以当作后期的耗时任务对待

&＃64;celery.task

def add(x,y):

    return x &＃43; y

       <2>启动celery服务

#cd /tmp

#celery -A tasks worker --loglevel&＃61;info

       <3>验证执行任务

       #导入模块,执行add函数,此处使用add.delay(3,4)而不是add(3,4)&＃xff0c;因为被celery封装了&＃xff0c;要异步执行需要额外使用add.delay(3,4)

       #需要注意&＃xff0c;如果把返回值赋值给一个变量&＃xff0c;那么原来的应用程序也会被阻塞&＃xff0c;需要等待异步任务返回的结果。因此&＃xff0c;实际使用中&＃xff0c;不需要把结果赋值。

#cd /tmp

#python

>>>from tasks import add

>>>add.delay(3,4)

       #celery服务的窗口会刷出任务的信息&＃xff0c;以及是否处理成功&＃xff0c;以及结果

       #将来只要在别的程序中引入tasks中的add函数&＃xff0c;就是异步的了&＃xff0c;是不是有点屌。。。。。

       <4>扩展知识&＃xff0c;指定队列名

       传入redis中的指定队列testq怎么玩&＃xff1f;&＃xff08;其他broker引擎也支持&＃xff09;

       启动celery服务的时候添加额外参数-Q &＃39;队列名&＃39;

#cd /tmp

#celery -A tasks.tasks worker --loglevel&＃61;info -Q &＃39;testq&＃39;

       跑任务的时候指定testq队列名

#cd /tmp

#python

>>>from tasks import add

>>>add.delay(3,4,queue&＃61;&＃39;testq&＃39;)

       <5>扩展知识&＃xff0c;指定开启的worker进程数(底层是调用的Python的multiprocessing模块中的Pool进程池思想来做)

       -c 5 开启5个worker进程来同时抢任务&＃xff0c;跑任务

#cd /tmp

#celery -A tasks.tasks worker --loglevel&＃61;info -c 5

       <6>扩展知识&＃xff0c;管理broker里面的数据&＃xff0c;查看任务状态&＃xff0c;以及任务的详细信息

       安装一个叫flower的webui&＃xff0c;提供任务查询&＃xff0c;worker生命管理&＃xff0c;以及路由管理等(底层是通过tornado框架封装的)

#pip install flower

#任意目录下执行都可以

#celery flower --port&＃61;5555 --broker&＃61;redis://localhost:6380/0

       #里面可以看到任务参数&＃xff0c;结果&＃xff0c;接受任务时间&＃xff0c;任务开始时间&＃xff0c;任务状态&＃xff0c;Started是任务进行中&＃xff0c;Success是任务跑完执行成功

    &＃xff08;4&＃xff09;Celery的使用二&＃xff08;项目方式&＃xff0c;也叫做Python包方式&＃xff0c;结构清晰&＃xff0c;低耦合&＃xff1b;相比纯脚本方式略复杂&＃xff0c;用不用由你&＃xff09;

    创建一个叫做proj的Python包&＃xff08;创建Python包的操作此处不做详细说明,tree /tmp/proj&＃xff09;

       <1>proj/celery.py

      #from __future__ import absolute_import据说添加此行可以低降低出错的概率哦&＃xff08;阿门保佑&＃xff1b;其实就是兼容Python版本的一个东东&＃xff09;

      #创建一个celery的实例&＃xff0c;名字叫做app&＃xff0c;传递进去的第一个参数是Python包的名字&＃xff0c;include加载任务文件&＃xff0c;config_from_object指定celery的配置文件(好吧&＃xff0c;看起来比单纯使用脚本方式麻烦点&＃xff0c;请继续往下看)

#!/usr/bin/env python

# -*- coding&＃61;utf-8 -*-

from __future__ import absolute_import

from celery import Celery

app &＃61; Celery(&＃39;proj&＃39;, include&＃61;[&＃39;proj.tasks&＃39;])

app.config_from_object(&＃39;proj.config&＃39;)

if __name__ &＃61;&＃61; &＃39;__main__&＃39;:

    app.start()

       <2>proj/config.py

       #配置文件里指定broker

#!/usr/bin/env python

# -*- coding:utf-8 -*-

from __future__ import absolute_import

BROKER_URL &＃61; &＃39;redis://127.0.0.1:6380/0&＃39;

       <3>proj/tasks.py

       #导入celery实例&＃xff0c;实例绑定任务

#!/usr/bin/env python

# -*- coding:utf-8 -*-

from __future__ import absolute_import

from proj.celery import app

&＃64;app.task

def add(x, y):

    return x &＃43; y

       <4>开启celery服务(特定目录指定包名字启动)

#cd /tmp/

#celery -A proj worker -l info

       <5>扩展功能&＃xff0c;指定队列名&＃xff0c;调整worker进程数&＃xff0c;页面管理celery同上&＃xff0c;不再做说明

 &＃xff08;5&＃xff09;Celery的使用三&＃xff08;django-celery模式;#反正我喜欢用这种&＃xff09;

       django调用celery跑异步任务&＃xff0c;常见场景有注册成功&＃xff0c;发送邮件可以异步来防止网络IO阻塞&＃xff0c;以及耗时间的任务&＃xff0c;例如需要去跑9000台IP的某些配置参数任务&＃xff0c;或者下发任务执行&＃xff0c;可能需要10几分钟才能跑完&＃xff0c;就可以WEB应用中使用这种异步方式

       <1>安装django-celery软件包

       #一定要注意celery的版本和django-celery的小版本要保持一致&＃xff0c;否则会有各种杂七杂八的小问题&＃xff08;都是泪.......&＃xff09;

#pip install celery&＃61;&＃61;3.1.17

#pip install django-celery&＃61;&＃61;3.1.17

       <2>创建celery必须的数据库表结构

#cd Python_20161203

#python manage.py migrate

       <3>django项目的settings.py文件中追加如下内容&＃xff1b;app呢是django项目里面的应用名字

       settings.py

import djcelery

djcelery.setup_loader()

BROKER_URL &＃61; &＃39;redis://127.0.0.1:6380/0&＃39;

CELERY_RESULT_BACKEND &＃61; &＃39;redis://127.0.0.1:6380/1&＃39;

CELERY_TASK_SERIALIZER &＃61; &＃39;json&＃39;

CELERY_RESULT_SERIALIZER &＃61; &＃39;json&＃39;

CELERY_ACCEPT_CONTENT &＃61; [&＃39;json&＃39;]

CELERY_IMPORTS &＃61; (&＃39;app.tasks&＃39;, )

CELERYBEAT_SCHEDULER &＃61; &＃39;djcelery.schedulers.DatabaseScheduler&＃39;

CELERYD_CONCURRENCY &＃61; 20

参数说明(可以根据自己的需求添加自己的参数)&＃xff1a;

CELERY_RESULT_BACKEND &＃61; "redis://127.0.0.1:6380/1&＃39;" #结果存储

CELERY_TASK_RESULT_EXPIRES &＃61; 1200 # celery任务执行结果的超时时间&＃xff0c;我的任务都不需要返回结果,只需要正确执行就行

CELERYD_CONCURRENCY &＃61; 20 # celery worker的并发数 也是命令行-c指定的数目,事实上实践发现并不是worker也多越好,保证任务不堆积,加上一定新增任务的预留就可以

CELERYD_PREFETCH_MULTIPLIER &＃61; 4 # celery worker 每次去redis取任务的数量&＃xff0c;我这里预取了4个慢慢执行,因为任务有长有短没有预取太多

CELERYD_MAX_TASKS_PER_CHILD &＃61; 200 # 每个worker执行了多少任务就会死掉&＃xff0c;我建议数量可以大一些&＃xff0c;比如200

CELERYBEAT_SCHEDULER &＃61; &＃39;djcelery.schedulers.DatabaseScheduler&＃39; # 这是使用了django-celery默认的数据库调度模型,任务执行周期都被存在你指定的orm数据库中

       <4>app/tasks.py&＃xff08;在django的app应用目录下创建tasks.py任务文件&＃xff0c;里面调用复杂的任务函数&＃xff09;

#!/usr/bin/env python

# -*- coding&＃61;utf-8 -*-

###############################

from __future__ import absolute_import

from celery import task

import time

#task装饰器封装了celery函数&＃xff0c;为耗时的操作

&＃64;task

def add(x,y):

    for i in range(30):

       print i

       time.sleep(1)

    return x &＃43; y

       <5>添加验证功能&＃xff0c;查看实际效果

       app/urls.py

urlpatterns &＃61; [

url(r&＃39;^celery_test/,views.celery_test),

]

       app/views.py

def celery_test(request):

    from tasks import add

    add.delay(4,8)

    return HttpResponse(&＃39;Celery testing666&＃39;)

       <6>开启djanog服务和celery服务&＃xff08;虽然耦合了&＃xff0c;但是还是需要额外开启&＃xff09;

#python manage.py runserver 0.0.0.0:8000

#另一个窗口开启celery服务

#python manage.py celery worker --loglevel&＃61;info

 

       <7>发送http的GET请求&＃xff0c;调用celery去执行异步任务&＃xff08;大功告成&＃xff09;

curl http://127.0.0.1:8000/index/celery_test/

        celery那端的屏幕输出如下&＃xff1a;

[2016-12-01 16:16:00,940: INFO/MainProcess] Received task: app.tasks.add[06a8d603-a7d3-4732-b8f3-ad010d531200]

[2016-12-01 16:16:00,941: WARNING/Worker-1] 0

[2016-12-01 16:16:01,943: WARNING/Worker-1] 1

[2016-12-01 16:16:02,945: WARNING/Worker-1] 2

[2016-12-01 16:16:03,947: WARNING/Worker-1] 3

[2016-12-01 16:16:04,948: WARNING/Worker-1] 4

[2016-12-01 16:16:05,950: WARNING/Worker-1] 5

[2016-12-01 16:16:06,952: WARNING/Worker-1] 6

[2016-12-01 16:16:07,954: WARNING/Worker-1] 7

[2016-12-01 16:16:08,955: WARNING/Worker-1] 8

[2016-12-01 16:16:09,957: WARNING/Worker-1] 9

[2016-12-01 16:16:10,958: WARNING/Worker-1] 10

[2016-12-01 16:16:11,959: WARNING/Worker-1] 11

[2016-12-01 16:16:12,961: WARNING/Worker-1] 12

[2016-12-01 16:16:13,962: WARNING/Worker-1] 13

[2016-12-01 16:16:14,964: WARNING/Worker-1] 14

[2016-12-01 16:16:15,964: WARNING/Worker-1] 15

[2016-12-01 16:16:16,966: WARNING/Worker-1] 16

[2016-12-01 16:16:17,968: WARNING/Worker-1] 17

[2016-12-01 16:16:18,969: WARNING/Worker-1] 18

[2016-12-01 16:16:19,971: WARNING/Worker-1] 19

[2016-12-01 16:16:20,973: WARNING/Worker-1] 20

[2016-12-01 16:16:21,974: WARNING/Worker-1] 21

[2016-12-01 16:16:22,976: WARNING/Worker-1] 22

[2016-12-01 16:16:23,978: WARNING/Worker-1] 23

[2016-12-01 16:16:24,979: WARNING/Worker-1] 24

[2016-12-01 16:16:25,981: WARNING/Worker-1] 25

[2016-12-01 16:16:26,982: WARNING/Worker-1] 26

[2016-12-01 16:16:27,984: WARNING/Worker-1] 27

[2016-12-01 16:16:28,986: WARNING/Worker-1] 28

[2016-12-01 16:16:29,987: WARNING/Worker-1] 29

[2016-12-01 16:16:30,990: INFO/MainProcess] Task app.tasks.add[06a8d603-a7d3-4732-b8f3-ad010d531200] succeeded in 30.049149203s: 12

    &＃xff08;6&＃xff09;Celery的使用四&＃xff08;celery模式;#反正我不喜欢用&＃xff0c;那索性不写了&＃xff0c;想学习的点击下面的链接去自助学习吧&＃xff0c;加油少年&＃xff09;

不安装django-celery版本使用方法