Celery从入门到放弃 - 简书
Celery从入门到放弃
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服务
#celery -A tasks worker --loglevel=info
&3&验证执行任务
#导入模块,执行add函数,此处使用add.delay(3,4)而不是add(3,4)，因为被celery封装了，要异步执行需要额外使用add.delay(3,4)
#需要注意，如果把返回值赋值给一个变量，那么原来的应用程序也会被阻塞，需要等待异步任务返回的结果。因此，实际使用中，不需要把结果赋值。
&&&from tasks import add
&&&add.delay(3,4)
#celery服务的窗口会刷出任务的信息，以及是否处理成功，以及结果
#将来只要在别的程序中引入tasks中的add函数，就是异步的了，是不是有点屌。。。。。
&4&扩展知识，指定队列名
传入redis中的指定队列testq怎么玩？（其他broker引擎也支持）
启动celery服务的时候添加额外参数-Q '队列名'
#celery -A tasks.tasks worker --loglevel=info -Q 'testq'
跑任务的时候指定testq队列名
&&&from tasks import add
&&&add.delay(3,4,queue='testq')
&5&扩展知识，指定开启的worker进程数(底层是调用的Python的multiprocessing模块中的Pool进程池思想来做)
-c 5 开启5个worker进程来同时抢任务，跑任务
#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
def add(x, y):
return x + y
&4&开启celery服务(特定目录指定包名字启动)
#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_
#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执行了多少任务就会死掉，我建议数量可以大一些，比如200CELERYBEAT_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函数，为耗时的操作
def add(x,y):
for i in range(30):
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那端的屏幕输出如下：[ 16:16:00,940: INFO/MainProcess] Received task: app.tasks.add[06a8d603-a7d3--ad010d531200][ 16:16:00,941: WARNING/Worker-1] 0[ 16:16:01,943: WARNING/Worker-1] 1[ 16:16:02,945: WARNING/Worker-1] 2[ 16:16:03,947: WARNING/Worker-1] 3[ 16:16:04,948: WARNING/Worker-1] 4[ 16:16:05,950: WARNING/Worker-1] 5[ 16:16:06,952: WARNING/Worker-1] 6[ 16:16:07,954: WARNING/Worker-1] 7[ 16:16:08,955: WARNING/Worker-1] 8[ 16:16:09,957: WARNING/Worker-1] 9[ 16:16:10,958: WARNING/Worker-1] 10[ 16:16:11,959: WARNING/Worker-1] 11[ 16:16:12,961: WARNING/Worker-1] 12[ 16:16:13,962: WARNING/Worker-1] 13[ 16:16:14,964: WARNING/Worker-1] 14[ 16:16:15,964: WARNING/Worker-1] 15[ 16:16:16,966: WARNING/Worker-1] 16[ 16:16:17,968: WARNING/Worker-1] 17[ 16:16:18,969: WARNING/Worker-1] 18[ 16:16:19,971: WARNING/Worker-1] 19[ 16:16:20,973: WARNING/Worker-1] 20[ 16:16:21,974: WARNING/Worker-1] 21[ 16:16:22,976: WARNING/Worker-1] 22[ 16:16:23,978: WARNING/Worker-1] 23[ 16:16:24,979: WARNING/Worker-1] 24[ 16:16:25,981: WARNING/Worker-1] 25[ 16:16:26,982: WARNING/Worker-1] 26[ 16:16:27,984: WARNING/Worker-1] 27[ 16:16:28,986: WARNING/Worker-1] 28[ 16:16:29,987: WARNING/Worker-1] 29[ 16:16:30,990: INFO/MainProcess] Task app.tasks.add[06a8d603-a7d3--ad010d531200] succeeded in 30.s: 12
（6）Celery的使用四（celery模式;#反正我不喜欢用，那索性不写了，想学习的点击下面的链接去自助学习吧，加油少年）
野生程序员任务调度利器：Celery - CSDN博客
任务调度利器：Celery
Celery简要介绍
Celery是Python开发的分布式任务调度模块，是一个异步的任务队列基于分布式消息传递。Celery本身不含消息服务，它使用第三方消息服务来传递任务，目前，Celery支持的消息服务有RabbitMQ、Redis甚至是数据库，当然Redis应该是最佳选择。
使用Redis作为Broker时，再安装一个celery-with-redis。
celery(芹菜)是一个异步任务队列/基于分布式消息传递的作业队列。
它侧重于实时操作，但对调度支持也很好。
celery用于生产系统每天处理数以百万计的任务。
celery是用Python编写的，但该协议可以在任何语言实现。它也可以用其他语言通过webhooks实现。
目前已知有php/ruby/nodejs的实现
为什么用Celery?
耗时久的事儿可以扔给 Worker处理，处理完可以触发子任务提醒
天然的并发能力(多进程/协程)!
非常方便添加 Worker来增强处理能力
Celery提供了Web方式的监控/报警，这样，我们就可以监控每个任务的情况了
出现错误可以自动处理/重试
Brokers: 提供队列服务，Celery支持的Brokers有:
RabbitMQ(推荐)
SQLAlchemy(MySQL/PostgreSQL/Sqlite/Oracle)
Amazon SQS等
Worker: 真正干活的，实际运行任务的节点。
开始 Celery 的第一步
选择Broker
在你正式开始使用 Celery之前，你需要选择、安装并运行一个 broker。
Broker 是一种负责接收、发送任务消息（task messages）的服务
你可以从以下几种 broker中选择一个：
RabbitMQ :[[/]]
功能完整、安全、饱经实践检验的 broker。如果对你而言，不丢失消息非常重要，RabbitMQ将是你最好的选择。
请查看 [[/技术分享/Celery/broker-installation]]以便获得更多关于
RabbitMQ安装和配置相关的信息。
Redis :[[http://redis.io/]]
也是一个功能完整的 broker，但电源故障或异常将导致数据丢失。
请查看 [[/技术分享/Celery/otherqueues-redis]]以便配置相关的信息。
不推荐使用数据库作为消息队列，但在非常小的应用中可以使用。Celery可以使用 SQLAlchemy和
DjangoORMS。请查看 [[/技术分享/Celery/otherqueues-sqlalchemy]]或
[[/技术分享/Celery/otherqueues-django]]。
更多其他选择
除了上面列出的以外，还有其他可以选择的传输实现，例如 CouchDB,Beanstalk, MongoDB, and SQS。请查阅 Kombu的文档以便获得更多信息。
创建一个简单“任务”（Task）
app = Celery(
&&& 'myapp',
&&& broker='amqp://guest@localhost//',
&&& #broker='redis://localhost:6379/0',
&&& # ## add result backend here ifneeded.
&&& backend='amqp',
&&& x +
运行 worker 服务器
celery -A tasks worker --loglevel=info
执行任务（task）
&&& result = add.delay(<span style="color:#, <span style="color:#)
这里是一些你可以如何处理结果的方法:
&&& result.ready() # 任务是否已经执行，如果已经执行完毕，返回&#20540;为 True。False
&&& result.result # 任务还没完成，尚无返回&#20540;。None
&&& result.get() # 等待，直到任务执行完毕并返回&#20540;。<span style="color:#
&&& result.result # 直接返回结果，不产生任何错误。<span style="color:#
&&& result.successful() # 任务是否成功执行，如果成功则返回&#20540; True。True
CELERY_RESULT_BACKEND =&amqp&#: 我们希望结果最多保存 5 分钟。&#: 注意，这个特性需要 2.1.1 以上版本 RabbitMQ 才能支持。#: 如果你使用的是早期版本的 RabbitMQ，请注释下面这行。
CELERY_TASK_RESULT_EXPIRES =<span style="color:#0
BROKER_URL = &amqp:''guest:guest@localhost :5672''& #连接 broker
CELERY_RESULT_BACKEND = &amqp& #用于存储元数据（metadata）和返回&#20540;（return values）的后端
CELERY_IMPORTS = (&tasks&, ) #列出 worker 需要 import 的模块
使用Redis作为Broker时，再安装一个celery-with-redis。
开始编写tasks.py：
# tasks.py
import time
from celery import Celery
celery = Celery('tasks', broker='redis://localhost:6379/0')
@celery.task
defsendmail(mail):
&&& print('sending mail to %s...' % mail['to'])
&&& time.sleep(<span style="color:#.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: <span style="color:#a0a9262-<span style="color:#58-<span style="color:#92-<span style="color:#81-b7bf0ea7483b&
可以看到，Celery的API设计真的非常简单。
然后，在Worker里就可以看到任务处理的消息：
[<span style="color:#13-<span style="color:#-<span style="color:#<span style="color:#:<span style="color:#:<span style="color:#,<span style="color:#3: WARNING/MainProcess] celery@MichaeliMac.local ready.
[<span style="color:#13-<span style="color:#-<span style="color:#<span style="color:#:<span style="color:#:<span style="color:#,<span style="color:#7: INFO/MainProcess] consumer: Connected to redis://localhost:<span style="color:#79/<span style="color:#.
[<span style="color:#13-<span style="color:#-<span style="color:#<span style="color:#:<span style="color:#:<span style="color:#,<span style="color:#8: INFO/MainProcess] Got task from broker: tasks.sendmail[<span style="color:#a0a9262-<span style="color:#58-<span style="color:#92-<span style="color:#81-b7bf0ea7483b]
[<span style="color:#13-<span style="color:#-<span style="color:#<span style="color:#:<span style="color:#:<span style="color:#,<span style="color:#5: WARNING/PoolWorker-<span style="color:#] sending mail to celery@python.org...
[<span style="color:#13-<span style="color:#-<span style="color:#<span style="color:#:<span style="color:#:<span style="color:#,<span style="color:#7: WARNING/PoolWorker-<span style="color:#] mail sent.
[<span style="color:#13-<span style="color:#-<span style="color:#<span style="color:#:<span style="color:#:<span style="color:#,<span style="color:#8: INFO/MainProcess] Task tasks.sendmail[<span style="color:#a0a9262-<span style="color:#58-<span style="color:#92-<span style="color:#81-b7bf0ea7483b] succeeded in<span style="color:#.s: None
要使celery找到相应的tasks有四种方法
1 在项目的celery中写tasks。如在djproj/djproj/celery.py中直接定义task
2 使用autodiscover 但是tasks必须定义在app目录下的名字为tasks.py的文件中，如：apps/app1/tasks.py 否则找不到 ，会报KeyError错误。
3 如果不使用autodiscover，可以在项目的celery中 import 相应的module，相当于在项目的celery中写了相应的task，如在celery.py中 import apps.app2.mytasks&
4 在settings.py中设置CELERY_IMPORTS = ('apps.app2.mytasks',) 写到module级&
第一,假设用户正发起一个request,
并等待request完成后返回.在这一request后面的view功能中,我们可能需要执行一段花费很长时间的程序任务,这一时间可能远远大于用户能忍受的范围.当这一任务并不需要立刻执行时,我们便可以使用Celery在后台执行,而不影响用户浏览网页.当有任务需要访问远程服务器完成时,我们往往都无法确定需要花费的时间.
第二则是定期执行某些任务.比如每小时需要检查一下天气预报,然后将数据储存到数据库中.我们可以编写这一任务,然后让Celery每小时执行一次.这样我们的web应用便能获取最新的天气预报信息.
本文已收录于以下专栏：
相关文章推荐
协程与多任务调度
时间&#160; 23:02:15
IT技术博客大学习
/it/article/7988?f=hot1
自从发了上次的文章使用celery之深入celery配置, 有一些网友再问我怎么让celery跑起来. 其实说来也是,celery在新手眼里真的是比较重量级,不好懂,今天先让他跑起来吧 本文大...
上次碰到js函数调度的问题，因为那时候不知道有event loop的概念……以为js中的函数也是抢占式的调用，然后又想起来上学期ucos中怪的碰撞检测有问题的情况……所以翻了下操作系统的书……然而之前...
本篇博文主要介绍在开发环境中的celery使用,请勿用于部署服务器.
许多Django应用需要执行异步任务, 以便不耽误http request的执行. 我们也可以选择许多方法来完成异步任务, 使用...
Celery用户指引–Workers启动、重启1.启动worker通过下面的命令在前台启动worker$ celery -A proj worker -l info通过下面的命令查看worker命令的...
启动worker
通过下面的命令在前台启动worker
$ celery -A proj worker -l info
通过下面的命令查看worker命令的所有相关选项
$ celery wor...
celery是Python开发的分布式异步任务调度系统，Celery支持的消息服务有rmq、redis等
以下代码使用的是redis作为消息队列，当然官网推荐生产环境使用rmq。
RabbitM...
操作系统任务调度问题。操作系统任务分为系统任务和用户任务两种。其中，系统任务的优先级 = 50且
&#160;&#160;&#160;&#160;&#160; 例如：task[] = {0, 30, 155, 1, 80, 300, 170, 4...
他的最新文章
讲师：吴岸城
您举报文章：
举报原因：
原文地址：
原因补充：
(最多只允许输入30个字)