celery学习笔记2

1.定义：

Celery是一个异步的任务队列（也叫做分布式任务队列）

2.工作结构

Celery分为3个部分

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

（2）broker部分负责任务消息的分发以及任务结果的存储，这部分任务主要由中间数据存储系统完成，比如消息队列服务器RabbitMQ、redis、Amazon SQS、MongoDB、IronMQ等或者关系型数据库，使用关系型数据库依赖sqlalchemy或者django的ORM

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

3.话不多说，用起来

（1）安装Celery(要安装celery3版本，4版本改动较大没测试)

pip install celery==3.1.17

（2）broker部分此处使用安装好的redis服务6380端口的db0作为消息队列，普通redis服务的安装此处不做介绍

（3）Celery的使用一（单独脚本调用，简单方便）

注：不考虑任务的结果存储情况

<1>/tmp/tasks.py（实际脚本中不要写中文注释）

#!/usr/bin/env python

# -*- coding=utf-8 -*-

from celery import Celery

from celery import platforms

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

platforms.C_FORCE_ROOT = True

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

celery = Celery(‘tasks‘, broker=‘redis://localhost:6380/0‘)

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

@celery.task

def add(x,y):

    return x + y

<2>启动celery服务

#cd /tmp

#celery -A tasks worker --loglevel=info

<3>验证执行任务

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

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

#cd /tmp

#python

>>>from tasks import add

>>>add.delay(3,4)

celery服务的窗口会刷出任务的信息，以及是否处理成功，以及结果

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

<4>扩展知识，指定队列名

传入redis中的指定队列testq怎么玩？（其他broker引擎也支持）

启动celery服务的时候添加额外参数-Q ‘队列名‘

#cd /tmp

#celery -A tasks.tasks worker --loglevel=info -Q ‘testq‘

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

#cd /tmp

#python

>>>from tasks import add

>>>add.delay(3,4,queue=‘testq‘)

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

-c 5 开启5个worker进程来同时抢任务，跑任务

#cd /tmp

#celery -A tasks.tasks worker --loglevel=info -c 5

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

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

#pip install flower

#任意目录下执行都可以

#celery flower --port=5555 --broker=redis://localhost:6380/0

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

（4）Celery的使用二（项目方式，也叫做Python包方式，结构清晰，低耦合；相比纯脚本方式略复杂，用不用由你）

创建一个叫做proj的Python包（创建Python包的操作此处不做详细说明,tree /tmp/proj）

<1>proj/celery.py

from __future__ import absolute_import 据说添加此行可以低降低出错的概率哦（阿门保佑；其实就是兼容Python版本的一个东东）

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

#!/usr/bin/env python

# -*- coding=utf-8 -*-

from __future__ import absolute_import

from celery import Celery

app = Celery(‘proj‘, include=[‘proj.tasks‘])

app.config_from_object(‘proj.config‘)

if __name__ == ‘__main__‘:

    app.start()

<2>proj/config.py

配置文件里指定broker

#!/usr/bin/env python

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

from __future__ import absolute_import

BROKER_URL = ‘redis://127.0.0.1:6380/0‘

<3>proj/tasks.py

导入celery实例，实例绑定任务

#!/usr/bin/env python

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

from __future__ import absolute_import

from proj.celery import app

@app.task

def add(x, y):

    return x + y

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

#cd /tmp/

#celery -A proj worker -l info

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

（5）Celery的使用三（django-celery模式;#反正我喜欢用这种）

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

<1>安装django-celery软件包

一定要注意celery的版本和django-celery的小版本要保持一致，否则会有各种杂七杂八的小问题（都是泪.......）

#pip install celery==3.1.17

#pip install django-celery==3.1.17

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

#cd Python_20161203

#python manage.py migrate

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

settings.py

import djcelery

djcelery.setup_loader()

BROKER_URL = ‘redis://127.0.0.1:6380/0‘

CELERY_RESULT_BACKEND = ‘redis://127.0.0.1:6380/1‘

CELERY_TASK_SERIALIZER = ‘json‘

CELERY_RESULT_SERIALIZER = ‘json‘

CELERY_ACCEPT_CONTENT = [‘json‘]

CELERY_IMPORTS = (‘app.tasks‘, )

CELERYBEAT_SCHEDULER = ‘djcelery.schedulers.DatabaseScheduler‘

CELERYD_CONCURRENCY = 20

参数说明(可以根据自己的需求添加自己的参数)：

CELERY_RESULT_BACKEND = "redis://127.0.0.1:6380/1‘" #结果存储

CELERY_TASK_RESULT_EXPIRES = 1200 # celery任务执行结果的超时时间，我的任务都不需要返回结果,只需要正确执行就行

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

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

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

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

<4>app/tasks.py（在django的app应用目录下创建tasks.py任务文件，里面调用复杂的任务函数）

#!/usr/bin/env python

# -*- coding=utf-8 -*-

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

from __future__ import absolute_import

from celery import task

import time

#task装饰器封装了celery函数，为耗时的操作

@task

def add(x,y):

    for i in range(30):

       print i

       time.sleep(1)

    return x + y

<5>添加验证功能，查看实际效果
app/urls.py

urlpatterns = [

  url(r‘^celery_test/,views.celery_test),

]

app/views.py

def celery_test(request):

    from tasks import add

    add.delay(4,8)

    return HttpResponse(‘Celery testing666‘)

<6>开启djanog服务和celery服务（虽然耦合了，但是还是需要额外开启）

#python manage.py runserver 0.0.0.0:8000

#另一个窗口开启celery服务

#python manage.py celery worker --loglevel=info

<7>发送http的GET请求，调用celery去执行异步任务（大功告成）

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

celery那端的屏幕输出如下：

[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

原文地址：https://www.cnblogs.com/zhouxinfei/p/8215607.html

时间： 2024-08-03 21:25:48

celery学习笔记2的相关文章

Swift学习笔记一

最近计划把Swift语言系统学习一下,然后将MagViewer用这种新语言重构一次,并且优化一下,这里记录一下Swift的学习笔记. Swift和Objective-C相比,在语法和书写形式上做了很多改进,面向开发者的体验更好了.比如: println("Hello, world!") 就可以打印出这个字符串,这条语句就是完整的一句,句末不需要添加分号,也不需要引入额外的头或者库文件比如标准输入输出等.在全局作用域内书写的代码就会作为程序的入口,因此不需要编写main函数了. 简单值

vector 学习笔记

vector 使用练习: /**************************************** * File Name: vector.cpp * Author: sky0917 * Created Time: 2014年04月27日 11:07:33 ****************************************/ #include <iostream> #include <vector> using namespace std; int main

Caliburn.Micro学习笔记(一)----引导类和命名匹配规则

Caliburn.Micro学习笔记(一)----引导类和命名匹配规则 用了几天时间看了一下开源框架Caliburn.Micro 这是他源码的地址http://caliburnmicro.codeplex.com/ 文档也写的很详细,自己在看它的文档和代码时写了一些demo和笔记,还有它实现的原理记录一下 学习Caliburn.Micro要有MEF和MVVM的基础 先说一下他的命名规则和引导类 以后我会把Caliburn.Micro的 Actions IResult,IHandle ICondu

jQuery学习笔记（一）：入门

jQuery学习笔记(一):入门 一.JQuery是什么 JQuery是什么?始终是萦绕在我心中的一个问题: 借鉴网上同学们的总结,可以从以下几个方面观察. 不使用JQuery时获取DOM文本的操作如下: 1 document.getElementById('info').value = 'Hello World!'; 使用JQuery时获取DOM文本操作如下: 1 $('#info').val('Hello World!'); 嗯,可以看出,使用JQuery的优势之一是可以使代码更加简练,使开

[原创]java WEB学习笔记93：Hibernate学习之路---Hibernate 缓存介绍，缓存级别，使用二级缓存的情况，二级缓存的架构集合缓存，二级缓存的并发策略，实现步骤，集合缓存，查询缓存，时间戳缓存

本博客的目的:①总结自己的学习过程,相当于学习笔记 ②将自己的经验分享给大家,相互学习,互相交流,不可商用 内容难免出现问题,欢迎指正,交流,探讨,可以留言,也可以通过以下方式联系. 本人互联网技术爱好者,互联网技术发烧友 微博:伊直都在0221 QQ:951226918 -----------------------------------------------------------------------------------------------------------------

Activiti 学习笔记记录（三）

上一篇:Activiti 学习笔记记录(二) 导读:上一篇学习了bpmn 画图的常用图形标记.那如何用它们组成一个可用文件呢? 我们知道 bpmn 其实是一个xml 文件

HTML&CSS基础学习笔记8-预格式文本

<pre>标签的主要作用是预格式化文本.被包围在 pre 标签中的文本通常会保留空格和换行符.而文本也会呈现为等宽字体. <pre>标签的一个常见应用就是用来表示计算机的源代码.当然你也可以在你需要在网页中预显示格式时使用它. 会使你的文本换行的标签(例如<h>.<p>)绝不能包含在 <pre> 所定义的块里.尽管有些浏览器会把段落结束标签解释为简单地换行,但是这种行为在所有浏览器上并不都是一样的. 更多学习内容,就在码芽网http://www.

java/android 设计模式学习笔记（14）---外观模式

这篇博客来介绍外观模式(Facade Pattern),外观模式也称为门面模式,它在开发过程中运用频率非常高,尤其是第三方 SDK 基本很大概率都会使用外观模式.通过一个外观类使得整个子系统只有一个统一的高层的接口,这样能够降低用户的使用成本,也对用户屏蔽了很多实现细节.当然,在我们的开发过程中,外观模式也是我们封装 API 的常用手段,例如网络模块.ImageLoader 模块等.其实我们在开发过程中可能已经使用过很多次外观模式,只是没有从理论层面去了解它. 转载请注明出处:http://bl

[原创]java WEB学习笔记48：其他的Servlet 监听器：域对象中属性的变更的事件监听器 （3 个），感知 Session 绑定的事件监听器（2个）

本博客为原创:综合 尚硅谷(http://www.atguigu.com)的系统教程(深表感谢)和 网络上的现有资源(博客,文档,图书等),资源的出处我会标明 本博客的目的:①总结自己的学习过程,相当于学习笔记 ②将自己的经验分享给大家,相互学习,互相交流,不可商用 内容难免出现问题,欢迎指正,交流,探讨,可以留言,也可以通过以下方式联系. 本人互联网技术爱好者,互联网技术发烧友 微博:伊直都在0221 QQ:951226918 ---------------------------------