任务调度利器：Celery - 廖雪峰的官方网站
任务调度利器：Celery
/ Reads: 41750
Celery是Python开发的分布式任务调度模块，今天抽空看了一下，果然接口简单，开发容易，5分钟就写出了一个异步发送邮件的服务。
Celery本身不含消息服务，它使用第三方消息服务来传递任务，目前，Celery支持的消息服务有RabbitMQ、Redis甚至是数据库，当然Redis应该是最佳选择。
安装Celery
用pip或easy_install安装：
$ sudo pip install Celery
$ sudo easy_install Celery
使用Redis作为Broker时，再安装一个celery-with-redis。
开始编写tasks.py：
# tasks.py
import time
from celery import Celery
celery = Celery('tasks', broker='redis://localhost:;)
@celery.task
def sendmail(mail):
print('sending mail to %s...' % mail['to'])
time.sleep(2.0)
print('mail sent.')
然后启动Celery处理任务：
$ celery -A tasks worker --loglevel=info
上面的命令行实际上启动的是Worker，如果要放到后台运行，可以扔给supervisor。
如何发送任务？非常简单：
&&& from tasks import sendmail
&&& sendmail.delay(dict(to='celery@python.org'))
&AsyncResult: 1a0a92-ea7483b&
可以看到，Celery的API设计真的非常简单。
然后，在Worker里就可以看到任务处理的消息：
[ 19:20:23,363: WARNING/MainProcess] celery@MichaeliMac.local ready.
[ 19:20:23,367: INFO/MainProcess] consumer: Connected to redis://localhost:6379/0.
[ 19:20:45,618: INFO/MainProcess] Got task from broker: tasks.sendmail[1a0a92-ea7483b]
[ 19:20:45,655: WARNING/PoolWorker-4] sending mail to celery@python.org...
[ 19:20:47,657: WARNING/PoolWorker-4] mail sent.
[ 19:20:47,658: INFO/MainProcess] Task tasks.sendmail[1a0a92-ea7483b] succeeded in 2.s: None
Celery默认设置就能满足基本要求。Worker以Pool模式启动，默认大小为CPU核心数量，缺省序列化机制是pickle，但可以指定为json。由于Python调用UNIX/Linux程序实在太容易，所以，用Celery作为异步任务框架非常合适。
Celery还有一些高级用法，比如把多个任务组合成一个原子任务等，还有一个完善的监控接口，以后有空再继续研究。
Make a Comment
Sign In to Make a Comment
You can sign in directly without register:
You need authorize to allow connect to your social passport for the first time.
WARNING: You are using an old browser that does not support HTML5.
Please choose a modern browser ( /
/ ) to get a good experience.Flask中使用celery队列处理执行时间较长的请求。
一. 安装celery
pip install celery flask
二. celery简介
Celery是个异步分布式任务队列
通过Celery在后台跑任务并不像线程那么简单，但是用Celery的话，能够是应用有较好的扩展性，因为Celery是个分布式架构，下面介绍Celery的三个核心组件：
1. 生产者(Celery client): 生产者发送消息，在Flask上工作时，生产者在Flask应用内运行
2. 消费者(Celert worker): 消费者用于处理后台任务。消费者可以是本地的也可以是远程的。我们可以在运行Flask的server上运行一个单一的消费者，当业务量上涨之后再去添加更多的消费者
3. 消息传递着(Celery broker): 生产者和消费者的信息交互使用的是消息队列，Celery支持若干方式的消息队列，其中最长用的是RabbitMQ和Redis, 我们在使用过程中使用的Redis
三. redis配置与使用
redis配置文件/etc/redis.conf
1.设置为后台启动
daemonize yes
2.redis端口设置
四. celery使用简介
1.Choosing a broker
最常用的broker包括: RabbitMQ 和 Redis, 我们使用Redis, Redis的安装及启动等查看第二部分
2.intall celery
pip install celery
3.Application
使用celery的第一步是创建一个application, 通常叫做'app'。具体的创建一个app的代码如下:
$ cat tasks.py
阅读(...) 评论()Celery--简介与设置 - 博客频道 - CSDN.NET
sicofield的专栏
分类：celery
Celery （芹菜）是基于Python开发的分布式任务队列。它支持使用任务队列的方式在分布的机器／进程／线程上执行任务调度。
Celery架构
Celery的架构由三部分组成，消息中间件（message broker），任务执行单元（worker）和任务执行结果存储（task result store）组成。
消息中间件
Celery本身不提供消息服务，但是可以方便的和第三方提供的消息中间件集成。包括，RabbitMQ, Redis, MongoDB (experimental), Amazon SQS (experimental),CouchDB (experimental), SQLAlchemy (experimental),Django ORM (experimental), IronMQ
任务执行单元
Worker是Celery提供的任务执行的单元，worker并发的运行在分布式的系统节点中。
任务结果存储
Task result store用来存储Worker执行的任务的结果，Celery支持以不同方式存储任务的结果，包括AMQP, Redis，memcached, MongoDB，SQLAlchemy, Django ORM，Apache Cassandra, IronCache
另外， Celery还支持不同的并发和序列化的手段。
Prefork, Eventlet, gevent, threads/single threaded
pickle, json, yaml, msgpack. zlib, bzip2 compression， Cryptographic message signing 等等
python安装celery
使用的是django-celery包：
pip install django-djcelery
配置djcelery
下面是djcelery的有关配置，定义在Django项目的settings模块内。
首先INSTALLED_APPS之中需要写入djcelery。
之后配置celery
说明：直接使用django做broker生产环境不建议，建议使用redis或者rabbitMQ
需要在settings.py之中载入celery的配置。
配置好了celery之后需要创建Celery所需要的数据表(django1.7)
python manage.py migrate
Celery使用
创建一个task
一个task就是一个Pyhton function。 但Celery需要知道这一function是task, 因此我们可以使用celery自带的装饰器decorator: @task. 在django app目录中创建taske.py:
from celery import task
def add(x, y):
return x + y
当settings.py中的djcelery.setup_loader()运行时, Celery便会查看所有INSTALLED_APPS中app目录中的tasks.py文件, 找到标记为task的function, 并将它们注册为celery task.
将function标注为task并不会妨碍他们的正常执行. 你还是可以像平时那样调用它:
z = add(1, 2).
celery之中你可以用 delay() 方法来调用任务。
让我们以一个简单的例子作为开始. 例如我们希望在用户发出request后异步执行该task, 马上返回response, 从而不阻塞该request, 使用户有一个流畅的访问过程. 那么, 我们可以使用.delay, 例如在在views.py的一个view中:
from myapp.tasks import add
add.delay(2, 2)
Celery会将task加入到queue中, 并马上返回. 而在一旁待命的worker看到该task后, 便会按照设定执行它, 并将他从queue中移除. 而worker则会执行以下代码:
import myapp.tasks.add
myapp.tasks.add(2, 2)
启动worker
正如之前说到的, 我们需要worker来执行task. 以下是在开发环境中的如何启动worker:
首先启动terminal, 如同开发django项目一样, 激活virtualenv, 切换到django项目目录. 然后启动django自带web服务器: python manage.py runserver.
然后启动worker:
python manage.py celery worker –loglevel=info
此时, worker将会在该terminal中运行, 并显示输出结果.
打开新的terminal, 激活virtualenv, 并切换到django项目目录:
$ python manage.py shell &&& from myapp.tasks import add &&& add.delay(2, 2)
此时, 你可以在worker窗口中看到worker执行该task:
[ 08:47:08,076: INFO/MainProcess] Got task from broker: myapp.tasks.add[e080e047-b2a2-43a7-af74-d7d9d98b02fc]
[ 08:47:08,299: INFO/MainProcess] Task myapp.tasks.add[e080e047-b2a2-43a7-af74-d7d9d98b02fc] succeeded in 0.s: 4
排名：第6200名
（13）（5）（12）（5）（18）（6）（19）（9）（6）（2）（4）（1）（13）（1）（2）（1）（3）（6）（3）（5）（2）（1）【转载】详解python多线程无法增加cpu使用率
原文地址：http://my.oschina.net/u/240562/blog/136651
GIL 与 Python 线程的纠葛
GIL 是什么东西？它对我们的 python 程序会产生什么样的影响？我们先来看一个问题。运行下面这段 python 程序，CPU
占用率是多少？
请勿在工作中模仿，危险:)
def dead_loop():
dead_loop()&
答案是什么呢，占用 100％ CPU？那是单核！还得是没有超线程的古董 CPU。在我的双核 CPU
上，这个死循环只会吃掉我一个核的工作负荷，也就是只占用 50％ CPU。那如何能让它在双核机器上占用 100％ 的 CPU
呢？答案很容易想到，用两个线程就行了，线程不正是并发分享 CPU
运算资源的吗。可惜答案虽然对了，但做起来可没那么简单。下面的程序在主线程之外又起了一个死循环的线程
import threading
def dead_loop():
新起一个死循环线程
t = threading.Thread(target=dead_loop)&
主线程也进入死循环
dead_loop()
按道理它应该能做到占用两个核的 CPU 资源，可是实际运行情况却是没有什么改变，还是只占了 50％ CPU 不到。这又是为什么呢？难道
python 线程不是操作系统的原生线程？打开 system monitor 一探究竟，这个占了 50% 的 python
进程确实是有两个线程在跑。那这两个死循环的线程为何不能占满双核 CPU 资源呢？其实幕后的黑手就是 GIL。
GIL 的迷思：痛并快乐着
GIL 的全程为 Global Interpreter Lock ，意即全局解释器锁。在 Python 语言的主流实现 CPython
中，GIL 是一个货真价实的全局线程锁，在解释器解释执行任何 Python 代码时，都需要先获得这把锁才行，在遇到 I/O
操作时会释放这把锁。如果是纯计算的程序，没有 I/O 操作，解释器会每隔 100
次操作就释放这把锁，让别的线程有机会执行（这个次数可以通过 sys.setcheckinterval 来调整）。所以虽然
CPython 的线程库直接封装操作系统的原生线程，但 CPython 进程做为一个整体，同一时间只会有一个获得了 GIL
的线程在跑，其它的线程都处于等待状态等着 GIL 的释放。这也就解释了我们上面的实验结果：虽然有两个死循环的线程，而且有两个物理
CPU 内核，但因为 GIL 的限制，两个线程只是做着分时切换，总的 CPU 占用率还略低于 50％。
看起来 python 很不给力啊。GIL 直接导致 CPython
不能利用物理多核的性能加速运算。那为什么会有这样的设计呢？我猜想应该还是历史遗留问题。多核 CPU 在 1990
年代还属于类科幻，Guido van Rossum 在创造 python 的时候，也想不到他的语言有一天会被用到很可能 1000＋
个核的 CPU
上面，一个全局锁搞定多线程安全在那个时代应该是最简单经济的设计了。简单而又能满足需求，那就是合适的设计（对设计来说，应该只有合适与否，而没有好与不好）。怪只怪硬件的发展实在太快了，摩尔定律给软件业的红利这么快就要到头了。短短
20 年不到，代码工人就不能指望仅仅靠升级 CPU
就能让老软件跑的更快了。在多核时代，编程的免费午餐没有了。如果程序不能用并发挤干每个核的运算性能，那就意谓着会被淘汰。对软件如此，对语言也是一样。那
Python 的对策呢？
Python 的应对很简单，以不变应万变。在最新的 python 3 中依然有 GIL。之所以不去掉，原因嘛，不外以下几点：
欲练神功，挥刀自宫：
CPython 的 GIL 本意是用来保护所有全局的解释器和环境状态变量的。如果去掉
GIL，就需要多个更细粒度的锁对解释器的众多全局状态进行保护。或者采用 Lock-Free
算法。无论哪一种，要做到多线程安全都会比单使用 GIL 一个锁要难的多。而且改动的对象还是有 20 年历史的 CPython
代码树，更不论有这么多第三方的扩展也在依赖 GIL。对 Python 社区来说，这不异于挥刀自宫，重新来过。
就算自宫，也未必成功：
有位牛人曾经做了一个验证用的 CPython，将 GIL
去掉，加入了更多的细粒度锁。但是经过实际的测试，对单线程程序来说，这个版本有很大的性能下降，只有在利用的物理 CPU
超过一定数目后，才会比 GIL 版本的性能好。这也难怪。单线程本来就不需要什么锁。单就锁管理本身来说，锁 GIL
这个粗粒度的锁肯定比管理众多细粒度的锁要快的多。而现在绝大部分的 python 程序都是单线程的。再者，从需求来说，使用 python
绝不是因为看中它的运算性能。就算能利用多核，它的性能也不可能和 C/C++ 比肩。费了大力气把 GIL
拿掉，反而让大部分的程序都变慢了，这不是南辕北辙吗。
难道 Python
这么优秀的语言真的仅仅因为改动困难和意义不大就放弃多核时代了吗？其实，不做改动最最重要的原因还在于：不用自宫，也一样能成功！
那除了切掉 GIL 外，果然还有方法让 Python 在多核时代活的滋润？让我们回到本文最初的那个问题：如何能让这个死循环的
Python 脚本在双核机器上占用 100％ 的 CPU？其实最简单的答案应该是：运行两个 python
死循环的程序！也就是说，用两个分别占满一个 CPU 内核的 python 进程来做到。确实，多进程也是利用多个 CPU
的好方法。只是进程间内存地址空间独立，互相协同通信要比多线程麻烦很多。有感于此，Python 在 2.6 里新引入了
multiprocessing 这个多进程标准库，让多进程的 python 程序编写简化到类似多线程的程度，大大减轻了 GIL
带来的不能利用多核的尴尬。
这还只是一个方法，如果不想用多进程这样重量级的解决方案，还有个更彻底的方案，放弃 Python，改用
C/C++。当然，你也不用做的这么绝，只需要把关键部分用 C/C++ 写成 Python 扩展，其它部分还是用 Python 来写，让
Python 的归 Python，C 的归 C。一般计算密集性的程序都会用 C 代码编写并通过扩展的方式集成到 Python
脚本里（如 NumPy 模块）。在扩展里就完全可以用 C 创建原生线程，而且不用锁 GIL，充分利用 CPU 的计算资源了。不过，写
Python 扩展总是让人觉得很复杂。好在 Python 还有另一种与 C 模块进行互通的机制 : ctypes
利用 ctypes 绕过 GIL
ctypes 与 Python 扩展不同，它可以让 Python 直接调用任意的 C 动态库的导出函数。你所要做的只是用 ctypes
写些 python 代码即可。最酷的是，ctypes 会在调用 C 函数前释放 GIL。所以，我们可以通过 ctypes 和 C
动态库来让 python 充分利用物理内核的计算能力。让我们来实际验证一下，这次我们用 C 写一个死循环函数
extern"C”&
void DeadLoop()&
用上面的 C 代码编译生成动态库 libdead_loop.so （Windows 上是 dead_loop.dll）
，接着就要利用 ctypes 来在 python 里 load 这个动态库，分别在主线程和新建线程里调用其中的
from ctypes import *&
from threading import Thread
cdll.LoadLibrary(“libdead_loop.so”)&
t = Thread(target=lib.DeadLoop)&
lib.DeadLoop()&
这回再看看 system monitor，Python 解释器进程有两个线程在跑，而且双核 CPU 全被占满了，ctypes
确实很给力！需要提醒的是，GIL 是被 ctypes 在调用 C 函数前释放的。但是 Python 解释器还是会在执行任意一段
Python 代码时锁 GIL 的。如果你使用 Python 的代码做为 C 函数的 callback，那么只要 Python 的
callback 方法被执行时，GIL 还是会跳出来的。比如下面的例子：
extern"C”&
typedef void Callback();&
void Call(Callback* callback)&
callback();
from ctypes import *&
from threading import Thread
def dead_loop():
lib = cdll.LoadLibrary(“libcall.so”)&
Callback = CFUNCTYPE(None)&
callback = Callback(dead_loop)
t = Thread(target=lib.Call,
args=(callback,))&
lib.Call(callback)&
注意这里与上个例子的不同之处，这次的死循环是发生在 Python 代码里 (DeadLoop 函数) 而 C 代码只是负责去调用这个
callback 而已。运行这个例子，你会发现 CPU 占用率还是只有 50％ 不到。GIL 又起作用了。
其实，从上面的例子，我们还能看出 ctypes 的一个应用，那就是用 Python 写自动化测试用例，通过 ctypes 直接调用 C
模块的接口来对这个模块进行黑盒测试，哪怕是有关该模块 C 接口的多线程安全方面的测试，ctypes 也一样能做到。
虽然 CPython 的线程库封装了操作系统的原生线程，但却因为 GIL 的存在导致多线程不能利用多个 CPU
内核的计算能力。好在现在 Python 有了易经筋（multiprocessing）, 吸星大法（C
语言扩展机制）和独孤九剑（ctypes），足以应付多核时代的挑战，GIL 切还是不切已经不重要了，不是吗。
引自http://zhuoqiang.me/python-thread-gil-and-ctypes.html
已投稿到：
以上网友发言只代表其个人观点，不代表新浪网的观点或立场。