Celery是一个专注于实时处理和任务调使用Celery - 炫6网
使用Celery
/ 作者：admin /
炫六网综合摘要：Celery是一个专注于实时处理和任务调 使用Celery，上一篇： 下一篇： 。Celery是一个专注于实时处理和任务调度的分布式任务队列。所谓任务就是消息，消息中的有效载荷中包含要执行任务需要的全部数据。 使用Celery的常见场景如下： Web应用。当用户触发的一个操作需要较长时间才能执行完成时，可以把它作为任务交给Celery去异步执
关键词：使用,Celery,Celery,一个,专注,于实时,实时,
Celery是一个专注于实时处理和任务调度的分布式任务队列。所谓任务就是消息，消息中的有效载荷中包含要执行任务需要的全部数据。使用Celery的常见场景如下：Web应用。当用户触发的一个操作需要较长时间才能执行完成时，可以把它作为任务交给Celery去异步执行，执行完再返回给用户。这段时间用户不需要等待，提高了的整体吞吐量和响应时间。定时任务。生产环境经常会跑一些定时任务。假如你有上千台的服务器、上千种任务，定时任务的管理很困难，Celery可以帮助我们快速在不同的机器设定不同种任务。其他可以异步执行的任务。为了充分提高网站性能，对于请求和响应之外的那些不要求必须同步完成的附加工作都可以异步完成。比如发送短信/邮件、推送消息、清理/设置缓存等。Celery还提供了如下的特性：方便地查看定时任务的执行情况，比如执行是否成功、当前状态、执行任务花费的时间等。可以使用功能齐备的管理后台或者命令行添加、更新、删除任务。方便把任务和配置管理相关联。可选多进程、Eventlet和Gevent三种模式并发执行。提供错误处理机制。提供多种任务原语，方便实现任务分组、拆分和调用链。支持多种消息代理和存储后端。Celery的架构Celery包含如下组件。Celery Beat：任务调度器，Beat进程会读取配置文件的内容，周期性地将配置中到期需要执行的任务发送给任务队列。Celery Worker：执行任务的消费者，通常会在多台服务器运行多个消费者来提高执行效率。Broker：消息代理，或者叫作消息中间件，接受任务生产者发送过来的任务消息，存进队列再按序分发给任务消费方（通常是消息队列或者数据库）。Producer：调用了Celery提供的API、函数或者装饰器而产生任务并交给任务队列处理的都是任务生产者。Result Backend：任务处理完后保存状态信息和结果，以供查询。Celery默认已支持Redis、RabbitMQ、MongoDB、Django ORM、SQLAlchemy等方式。Celery的架构图如图9.3所示。图9-3 Celery的架构图选择消息代理Celery目前支持RabbitMQ、Redis、MongoDB、Beanstalk、SQLAlchemy、Zookeeper等作为消息代理，但适用于生产环境的只有RabbitMQ和Redis，至于其他的方式，一是支持有限，二是可能得不到更好的技术支持。Celery官方推荐的是RabbitMQ，Celery的作者Ask Solem Hoel最初在VMware就是为RabbitMQ工作的，Celery最初的设计就是基于RabbitMQ，所以使用RabbitMQ会非常稳定，成功案例很多。如果使用Redis，则需要能接受发生突然断电之类的问题造成Redis突然终止后的数据丢失等后果。Celery序列化在客户端和消费者之间传输数据需要序列化和反序列化，Celery支持如表9.2所示的序列化方案。表9.2 Celery支持的序列化方案安装配置Celery为了提供更高的性能，我们选择如下方案：选择RabbitMQ作为消息代理。RabbitMQ的Python客户端选择librabbitmq这个C库。选择Msgpack做序列化。选择Redis做结果存储。下面先安装它们。Celery提供bundles的方式，也就是安装Celery的同时可以一起安装多种依赖：pip install "celery[librabbitmq,redis,msgpack]" bundles的原理是在setup.py的setup函数中添加extras_require。从一个简单的例子开始先演示一个简单的项目让Celery运行起来。项目的目录结构如下：? tree chapter9/section3/proj├── celeryconfig.py├── celery.py├── __init__.py└── tasks.py先看一下主程序celery.py：解析一下这个程序：“from __future__ import absolute_import”是拒绝隐式引入，因为celery.py的名字和celery的包名冲突，需要使用这条语句让程序正确地运行。app是Celery类的实例，创建的时候添加了proj.tasks这个模块，也就是包含了proj/tasks.py这个文件。把Celery配置存放进proj/celeryconfig.py文件，使用app.config_from_object加载配置。看一下存放任务函数的文件tasks.py：tasks.py只有一个任务函数add，让它生效的最直接的方法就是添加app.task这个装饰器。看一下我们的配置文件celeryconfig.py：这个例子中没有任务调度相关的内容， 所以只需要启动消费者：? cd ~/web_develop/chapter9/section3? celery -A proj worker -l info-A参数默认会寻找proj.celery这个模块，其实使用celery作为模块文件名字不怎么合理。可以使用其他名字。举个例子，假如是proj/app.py，可以使用如下命令启动：? celery -A proj.app worker -l info如果看到如下的启动信息，就说明worker服务运行起来了：上述信息提供了一些有帮助的内容，如消息代理和存储结果的地址、并发数量、任务列表、交换类型等。在对Celery不熟悉的时候可以通过如上信息判断设置和修改是否已生效。现在开启另外一个终端，用IPython调用add函数：可以看到worker的终端上显示执行了任务：通过IPython触发的任务就完成了。任务的结果都需要根据上面提到的task_id获得，我们还可以用如下两种方式随时找到这个结果：指定队列Celery非常容易设置和运行，通常它会使用默认的名为celery的队列（可以通过CELERY_DEFAULT_QUEUE修改）用来存放任务。我们可以使用优先级不同的队列来确保高优先级的任务不需要等待就得到响应。基于proj目录下的源码，我们创建一个projq目录，并对projq/celeryconfig.py添加如下配置：现在用指定队列的方式启动消费者进程：? celery -A projq worker -Q web_tasks -l info上述worker只会执行web_tasks中的任务，我们可以合理安排消费者数量，让web_tasks中任务的优先级更高。使用任务调度之前的例子都是由发布者触发的，本节展示一下使用Celery的Beat进程自动生成任务。基于proj目录下的源码，创建一个projb目录，对projb/celeryconfig.py添加如下配置：CELERYBEAT_SCHEDULE中指定了tasks.add这个任务每10秒跑一次，执行的时候的参数是16和16。启动Beat程序：? celery beat -A projb然后启动Worker进程：? celery -A projb worker -l info之后可以看到每10秒都会自动执行一次tasks.add。Beat和Worker进程可以一并启动：? celery -B -A projb worker -l info使用Django可以通过django-celery实现在管理后台创建、删除、更新任务，是因为它使用了自定义的调度类djcelery.schedulers.DatabaseScheduler，我们可以参考它实现Flask或者其他Web框架的管理后台来完成同样的功能。使用自定义调度类还可以实现动态添加任务。任务绑定、记录日志和重试任务绑定、记录日志和重试是Celery常用的3个高级属性。现在修改proj/tasks.py文件，添加div函数用于演示：当使用bind = True后，函数的参数发生变化，多出了参数self（第一个参数），相当于把div变成了一个已绑定的方法，通过self可以获得任务的上下文。在IPython中调用div：In : from proj.tasks import divIn : r = div.delay(2, 1)可以看到如下执行信息：[ 15:50:31,853: INFO/Worker-1] proj.tasks.div[1da82fb8-20de-4d5a-9b48-045da6db0cda]: Executing task id 1da82fb8-20de-4d5a-9b48-045da6db0cda, args: [2, 1] kwargs: {}换成能造成异常的参数：In : r = div.delay(2, 0)可以发现每5秒就会重试一次，一共重试3次（默认重复3次），然后抛出异常。在Flask应用中使用Celery在Web应用中，用户请求页面发布任务，交由Celery后端处理。一种是把任务发布然后请求继续进行，响应不需要获知任务的执行情况；另外一种是需要实时把任务的执行过程反馈到用户的浏览器上。本节将演示这两种任务的处理方式。Socket.IO是一个支持WebSocket协议，面向实时Web应用的JavaScript库。实现了浏览器与服务器之间的双向通信。它有两个部分：在浏览器中运行的客户端库。这个库由Socket.IO官方提供。Python实现的服务端库。为了简化代码，我们使用Flask-SocketIO这个扩展，首先安装它：? pip install Flask-SocketIOCelery默认使用多进程的方式运行Worker进程，这个例子将使用eventlet的方法运行Worker以及SocketIO。首先需要引入依赖包和设置（celery_socketio.py）：接下来编写Celery任务：然后编写视图函数：最后，我们设置启动应用的方式，和之前的Flask用法略有不同：if __name__ == '__main__': socketio.run(app, host='0.0.0.0', port=9000, debug=True)创建一个简单的模板index.html，它带有很少的样式，用jQuery操作DOM。为了减少依赖，socket.io.js引用了外部CDN的地址：script type="text/javascript" src="//cdn.socket.io/socket.io-1.4.5.js"/script先看一下模板的主体：再看一下JavaScript事件部分：上面这段代码表示当页面加载完成之后，我们会创建Websocket链接，并且给id为background的DOM添加一个只要点击就会访问视图函数名为start_background_task（也就是访问/task）的事件。现在运行Flask应用和Celery：? cd ~/web_develop/chapter9/section3? python celery_socketio.py? celery -A celery_socketio.celery -P eventlet worker -l info访问“http://localhost:9000/async”可以感受到这个请求是直接返回的，sleep 5秒的操作并没有影响到页面响应时间。访问首页“http://localhost:9000/”，单击Execute按钮，可以看到红色框里面的内容会更新，执行完毕后如图9.4所示。图9.4 红色框里面的内容会更新本文节选自《Python Web开发实战》一书董伟明 著电子工业出版社出版
使用Celery
通过E-mail将您的想法和建议发给我们
我要投稿：
商务合作QQ：基于Django与Celery实现异步队列任务 - Python - 伯乐在线
& 基于Django与Celery实现异步队列任务
在运营系统中经常用到异步方式来处理我们的任务，比如将业务上线流程串成任务再写入队列，通过后台作业节点去调度执行。比较典型的案例为腾讯的蓝鲸、织云、云智慧等平台。本译文结合Django+Celery+Redis实现一个定期从Flickr 获取图片并展示的简单案例，方便大家理解实现异步对列任务的过程。
刚接触django的时候，我经历过的最让人沮丧的事情是需要定期运行一段代码。我写了一个需要每天上午12点执行一个动作的不错的函数。很简单是不是？错了。事实证明，这对我来说是一个巨大的困难点，因为，那时我使用Cpane类型的虚拟主机管理系统，它能专门提供一个很友好，很方便的图形用户界面来设置cron作业。
经过反复研究，我发现了一个很好的解决方案 – Celery，一个用于在后台运行任务的强大的异步作业队列。但是，这也导致了其它的问题，因为我无法找到一系列简单的指令将celery集成到Django项目中。
当然，我最终还是设法成功搞定了它 – 这正是本文将介绍的内容：如何将celery集成到一个Django项目，创建周期性任务。
该项目利用Python3.4，Django的1.8.2，celery3.1.18和Redis3.0.2.
由于大篇幅的文字，为了您的方便，请参阅下表中的每一步的简要信息，并获取相关的代码。
集成Celery和Django
Celery任务
添加基本的Celery任务
周期性任务
添加周期性任务
本地运行我们的应用程序
远程运行我们的应用程序
二、什么是Celery
“Celery是一个异步任务队列/基于分布式消息传递的作业队列。它侧重于实时操作，但对调度的支持也很好。”本文，我们将重点讲解周期性执行任务的调度特点。
为什么这一点有用呢？
回想一下你不得不在将来运行某一特定任务的经历。也许你需要每隔一小时访问一个API。或者，也许你需要在这一天结束时发送一批电子邮件。不论任务大小，Celery都可以使得调度周期性任务变的很容易。
你永远不希望终端用户等待那些不必要的页面加载或动作执行完成。如果你的应用程序工作流的一部分是一个需要很长时间的程序，当资源可用时，你就可以使用Celery在后台执行这段程序，从而使你的应用程序可以继续响应客户端的请求。这样可以使任务在应用程序的环境之外运行。
三、构建项目
在深入了解Celery之前，先从Github库中获取开始项目。确保激活一个虚拟的环境，安装必要的软件，并运行迁移。然后启动服务器，通过你的浏览器导航到http://localhost:8000/。你应当能看到‘恭喜你的第一个Django页面’。完成后，关闭服务器。
接下来，我们开始安装celery。
$ pip install celery==3.1.18
$ pip freeze & requirements.txt
$ pip install celery==3.1.18 $ pip freeze & requirements.txt
现在，我们通过简单的三步将celery集成到django项目中。
步骤一：创建celery.py
在“picha“目录下，创建celery.py，代码如下：
from __future__ import absolute_import
from celery import Celery
from django.conf import settings
# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'picha.settings')
app = Celery('picha')
# Using a string here means the worker will not have to
# pickle the object when using Windows.
app.config_from_object('django.conf:settings')
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
@app.task(bind=True)
def debug_task(self):
print('Request: {0!r}'.format(self.request))
1234567891011121314151617
from __future__ import absolute_import import os from celery import Celery from django.conf import settings &# set the default Django settings module for the 'celery' program. os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'picha.settings') app = Celery('picha') &# Using a string here means the worker will not have to # pickle the object when using Windows. app.config_from_object('django.conf:settings') app.autodiscover_tasks(lambda: settings.INSTALLED_APPS) &@app.task(bind=True) def debug_task(self): &&&&print('Request: {0!r}'.format(self.request))
请注意代码中的注释。
步骤二：引入celery应用
为了确保在django启动时加载了celery应用，在settings.py旁边新建__init__.py，并添加以下代码到__init__.py中。
from __future__ import absolute_import
# This will make sure the app is always imported when
# Django starts so that shared_task will use this app.
from .celery import app as celery_app
from __future__ import absolute_import &# This will make sure the app is always imported when # Django starts so that shared_task will use this app. from .celery import app as celery_app
完成以上步骤后，你的项目目录应该是这样的：
├── manage.py
├── picha
├── __init__.py
├── celery.py
├── settings.py
├── urls.py
└── wsgi.py
└── requirements.txt
├── manage.py ├── picha │&& ├── __init__.py │&& ├── celery.py │&& ├── settings.py │&& ├── urls.py │&& └── wsgi.py └── requirements.txt
步骤三：安装 Redis作为Celery的“中间件”
Celery使用中间件在django项目与celery监控者之间传递消息。在本教程中，我们使用redis作为消息中间代理。
首先，从官方下载页面或通过brew（BREW安装Redis）安装Redis，然后打开你的终端上，在一个新的终端窗口，启动服务器：
$ redis-server
$ redis-server
你可以通过在终端中输入如下命令测试Redis是否正常工作。
$ redis-cli ping
$ redis-cli ping
Redis应该回复PONG – 试试吧！
一旦Redis正常启动了，把下面的代码添加到你的settings.py文件中：
# CELERY STUFF
BROKER_URL = 'redis://localhost:6379'
CELERY_RESULT_BACKEND = 'redis://localhost:6379'
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Africa/Nairobi'
# CELERY STUFF BROKER_URL = 'redis://localhost:6379' CELERY_RESULT_BACKEND = 'redis://localhost:6379' CELERY_ACCEPT_CONTENT = ['application/json'] CELERY_TASK_SERIALIZER = 'json' CELERY_RESULT_SERIALIZER = 'json' CELERY_TIMEZONE = 'Africa/Nairobi'
你还需要添加Redis的作为Django项目的依赖：
$ pip install redis==2.10.3
$ pip freeze & requirements.txt
$ pip install redis==2.10.3 $ pip freeze & requirements.txt
就是这样了！你现在应该能够在Django中使用Celery。有关设置Celery与Django的更多信息，请查看官方Celery文档。
在继续下面步骤之前，让我们进行一些完整性检查，以确保一切都是正常的。
测试Celery worker已准备好接收任务：
$ celery -A picha worker -l info
[ 14:07:07,398: INFO/MainProcess] Connected to redis://localhost:6379//
[ 14:07:07,410: INFO/MainProcess] mingle: searching for neighbors
[ 14:07:08,419: INFO/MainProcess] mingle: all alone
$ celery -A picha worker -l info ... [2015-07-07 14:07:07,398: INFO/MainProcess] Connected to redis://localhost:6379// [2015-07-07 14:07:07,410: INFO/MainProcess] mingle: searching for neighbors [2015-07-07 14:07:08,419: INFO/MainProcess] mingle: all alone
使用CTRL-C杀死该段程序。现在，测试Celery任务调度程序是否已经准备好：
$ celery -A picha beat -l info
[ 14:08:23,054: INFO/MainProcess] beat: Starting...
$ celery -A picha beat -l info ... [2015-07-07 14:08:23,054: INFO/MainProcess] beat: Starting...
在上述完成时再次终止该进程。
1、Celery任务
Celery利用celery调用的常规Python函数作为任务。
例如，让我们把这个基本函数变为celery的任务：
def add(x, y):
return x + y
def add(x, y): &&&&return x + y
首先，添加一个装饰器。
from celery.decorators import task
@task(name="sum_two_numbers")
def add(x, y):
return x + y
from celery.decorators import task &@task(name="sum_two_numbers") def add(x, y): &&&&return x + y
然后你可以通过以下方式利用celery异步运行该任务：
add.delay(7, 8)
add.delay(7, 8)
很简单，对不对？
所以，这对于解决类似你要加载一个网页，而不需要用户等待一些后台程序的完成这些类型的任务来说是非常完美的。
让我们来看一个例子…
让我们再回到Django项目的版本3，它包括一个接受来自用户的反馈的应用程序，人们形象地称之为反馈：
├── feedback
├── __init__.py
├── admin.py
├── emails.py
├── forms.py
├── models.py
├── tests.py
└── views.py
├── manage.py
├── picha
├── __init__.py
├── celery.py
├── settings.py
├── urls.py
└── wsgi.py
├── requirements.txt
└── templates
├── base.html
└── feedback
├── contact.html
└── email
├── feedback_email_body.txt
└── feedback_email_subject.txt
1234567891011121314151617181920212223
├── feedback&&│&& ├── __init__.py&&│&& ├── admin.py&&│&& ├── emails.py&&│&& ├── forms.py&&│&& ├── models.py&&│&& ├── tests.py&&│&& └── views.py&&├── manage.py&&├── picha&&│&& ├── __init__.py&&│&& ├── celery.py&&│&& ├── settings.py&&│&& ├── urls.py&&│&& └── wsgi.py&&├── requirements.txt&&└── templates&&&&&&├── base.html&&&&&&└── feedback&&&&&&&&&&├── contact.html&&&&&&&&&&└── email&&&&&&&&&&&&&&├── feedback_email_body.txt&&&&&&&&&&&&&&└── feedback_email_subject.txt
安装新的必要软件，启动应用程序，并导航到http://localhost:8000/feedback/。你应该看到如下结果：
让我们连接celery任务。
2、添加任务
基本上，用户提交反馈表后，我们希望让他继续以他舒服的方式往下进行，而我们在后台进行处理反馈，发送电子邮件等等。
要做到这一点，首先添加一个叫tasks.py的文件到“feedback”目录：
from celery.decorators import task
from celery.utils.log import get_task_logger
from feedback.emails import send_feedback_email
logger = get_task_logger(__name__)
@task(name="send_feedback_email_task")
def send_feedback_email_task(email, message):
"""sends an email when feedback form is filled successfully"""
("Sent feedback email")
return send_feedback_email(email, message)
123456789101112
from celery.decorators import task&&from celery.utils.log import get_task_logger&&&from feedback.emails import send_feedback_email&&&logger = get_task_logger(__name__)&&&@task(name="send_feedback_email_task")&&def send_feedback_email_task(email, message):&&&&&&"""sends an email when feedback form is filled successfully"""&&&&&&logger.info("Sent feedback email")&&&&&&return send_feedback_email(email, message)
然后按照如下内容更新forms.py：
from django import forms
from feedback.tasks import send_feedback_email_task
class FeedbackForm(forms.Form):
email = forms.EmailField(label="Email Address")
message = forms.CharField(
label="Message", widget=forms.Textarea(attrs={'rows': 5}))
honeypot = forms.CharField(widget=forms.HiddenInput(), required=False)
def send_email(self):
# try to trick spammers by checking whether the honeypot field is
# not super complicated/effective but it works
if self.cleaned_data['honeypot']:
return False
send_feedback_email_task.delay(
self.cleaned_data['email'], self.cleaned_data['message'])
12345678910111213141516
from django import forms&&from feedback.tasks import send_feedback_email_task&&&class FeedbackForm(forms.Form):&&&&&&email = forms.EmailField(label="Email Address")&&&&&&message = forms.CharField(&&&&&&&&&&label="Message", widget=forms.Textarea(attrs={'rows': 5}))&&&&&&honeypot = forms.CharField(widget=forms.HiddenInput(), required=False)&&&&&&&def send_email(self):&&&&&&&&&&# try to trick spammers by checking whether the honeypot field is&&&&&&&&&&# not super complicated/effective but it works&&&&&&&&&&if self.cleaned_data['honeypot']:&&&&&&&&&&&&&&return False&&&&&&&&&&send_feedback_email_task.delay(&&&&&&&&&&&&&&self.cleaned_data['email'], self.cleaned_data['message'])
大体上，send_feedback_email_task.delay(email, message)的函数过程，并发送反馈电子邮件等都是在用户继续使用该网站的同时作为后台进程运行。
注：在views.py中的success_url被设置为将用户重定向到/ 目录，这个目录还不存在。我们会在下一节设置这个终点启动。
3、周期任务
通常情况下，你经常需要安排一个任务在特定的时间运行 – 例如，一个web scraper 可能需要每天都运行。这样的任务，被称为周期性任务，很容易建立利用celery启动。
celery使用“celery beat”来安排定期任务。celery beat定期运行任务，然后由celery worker执行任务。
例如，下面的任务计划每15分钟运行一次：
from celery.task.schedules import crontab
from celery.decorators import periodic_task
@periodic_task(run_every=(crontab(minute='*/15')), name="some_task", ignore_result=True)
def some_task():
# do something
from celery.task.schedules import crontab&&from celery.decorators import periodic_task&&&@periodic_task(run_every=(crontab(minute='*/15')), name="some_task", ignore_result=True)&&def some_task():&&&&&&# do something
让我们通过往Django项目中添加功能来看一个更强大的例子。
回到Django项目版本4，它包括另一个新的应用程序，叫做photos，这个应用程序使用 Flickr API获取新照片用来显示在网站：
├── feedback
├── __init__.py
├── admin.py
├── emails.py
├── forms.py
├── models.py
├── tasks.py
├── tests.py
└── views.py
├── manage.py
├── photos
├── __init__.py
├── admin.py
├── models.py
├── settings.py
├── tests.py
├── utils.py
└── views.py
├── picha
├── __init__.py
├── celery.py
├── settings.py
├── urls.py
└── wsgi.py
├── requirements.txt
└── templates
├── base.html
├── feedback
├── contact.html
└── email
├── feedback_email_body.txt
└── feedback_email_subject.txt
└── photos
└── photo_list.html
12345678910111213141516171819202122232425262728293031323334
├── feedback&&│&& ├── __init__.py&&│&& ├── admin.py&&│&& ├── emails.py&&│&& ├── forms.py&&│&& ├── models.py&&│&& ├── tasks.py&&│&& ├── tests.py&&│&& └── views.py&&├── manage.py&&├── photos&&│&& ├── __init__.py&&│&& ├── admin.py&&│&& ├── models.py&&│&& ├── settings.py&&│&& ├── tests.py&&│&& ├── utils.py&&│&& └── views.py&&├── picha&&│&& ├── __init__.py&&│&& ├── celery.py&&│&& ├── settings.py&&│&& ├── urls.py&&│&& └── wsgi.py&&├── requirements.txt&&└── templates&&&&&&├── base.html&&&&&&├── feedback&&&&&&│&& ├── contact.html&&&&&&│&& └── email&&&&&&│&&&&&& ├── feedback_email_body.txt&&&&&&│&&&&&& └── feedback_email_subject.txt&&&&&&└── photos&&&&&&&&&&└── photo_list.html
安装新的必要软件，运行迁移，然后启动服务器，以确保一切都是好的。重新测试反馈表。这次，它应该重定向好了。
下一步是什么？
既然我们需要周期性的调用Flickr API，以获取更多的照片添加到我们的网站，我们可以添加一个celery任务。
4、添加任务
往photos应用中添加一个tasks.py。
from celery.task.schedules import crontab
from celery.decorators import periodic_task
from celery.utils.log import get_task_logger
from photos.utils import save_latest_flickr_image
logger = get_task_logger(__name__)
@periodic_task(
run_every=(crontab(minute='*/15')),
name="task_save_latest_flickr_image",
ignore_result=True
def task_save_latest_flickr_image():
Saves latest image from Flickr
save_latest_flickr_image()
("Saved image from Flickr")
12345678910111213141516171819
from celery.task.schedules import crontab&&from celery.decorators import periodic_task&&from celery.utils.log import get_task_logger&&&from photos.utils import save_latest_flickr_image&&&logger = get_task_logger(__name__)&&&@periodic_task(&&&&&&run_every=(crontab(minute='*/15')),&&&&&&name="task_save_latest_flickr_image",&&&&&&ignore_result=True&&)&&def task_save_latest_flickr_image():&&&&&&""" &&&&Saves latest image from Flickr &&&&"""&&&&&&save_latest_flickr_image()&&&&&&logger.info("Saved image from Flickr")
在这里，我们通过在一个task中包装这个函数，来实现每15分钟运行一次save_latest_flickr_image()函数。该@periodic_task装饰器抽象出代码来运行celery任务，使得tasks.py干净，易于阅读！
5、本地运行
准备开始运行了？
在Django应用程序和Redis运行的前提下，打开两个新的终端窗口/标签。在每一个新的窗口中，导航到你的项目目录，激活你的虚拟环境，然后运行下面的命令（每个窗口一个）：
$ celery -A picha worker -l info
$ celery -A picha beat -l info
$ celery -A picha worker -l info $ celery -A picha beat -l info
当你访问http://127.0.0.1:8000/ 网址的时候，你现在应该能看到一个图片。我们的应用程序每15分钟从Flickr 获取一张图片。
通过photos/tasks.py查看代码。点击“Feedback”按钮发送一些反馈意见：
以上是通过celery任务运行的。更多的请查看feedback/tasks.py。
就这样，你成功的启动并运行了 Picha项目！
当你本地开发Django项目时，这是一个很好的测试，但是当你需要部署到生产环境- 就像 DigitalOcean时，就不那么合适了。为此，建议你通过使用Supervisor在后台作为一个守护进程运行celery worker和调度器。
6、远程运行
安装很简单。从版本库中获取版本5（如果你还没有的话）。然后，SSH到远程服务器，并运行：
$ sudo apt-get install supervisor
$ sudo apt-get install supervisor
然后，通过在远程服务器上“/etc/supervisor/conf.d/” 目录下添加配置文件来告知Supervisor celery的workers。在我们的例子中，我们需要两个这样的配置文件 – 一个用于Celery worker，一个是Celery scheduler。
在本地，在项目的根目录下创建一个“supervisor”的文件夹，然后添加下面的文件。
Celery Worker: picha_celery.conf
; ==================================
celery worker supervisor example
; ==================================
; the name of your supervisord program
[program:pichacelery]
; Set full path to celery program if using virtualenv
command=/home/mosh/.virtualenvs/picha/bin/celery worker -A picha --loglevel=INFO
; The directory to your Django project
directory=/home/mosh/sites/picha
; If supervisord is run as the root user, switch users to this UNIX user account
; before doing any processing.
; Supervisor will start as many instances of this program as named by numprocs
numprocs=1
; Put process stdout output in this file
stdout_logfile=/var/log/celery/picha_worker.log
; Put process stderr output in this file
stderr_logfile=/var/log/celery/picha_worker.log
; If true, this program will start automatically when supervisord is started
autostart=true
; May be one of false, unexpected, or true. If false, the process will never
; be autorestarted. If unexpected, the process will be restart when the program
; exits with an exit code that is not one of the exit codes associated with this
; process’ configuration (see exitcodes). If true, the process will be
; unconditionally restarted when it exits, without regard to its exit code.
autorestart=true
; The total number of seconds which the program needs to stay running after
; a startup to consider the start successful.
startsecs=10
; Need to wait for currently executing tasks to finish at shutdown.
; Increase this if you have very long running tasks.
stopwaitsecs = 600
; When resorting to send SIGKILL to the program to terminate it
; send SIGKILL to its whole process group instead,
; taking care of its children as well.
killasgroup=true
; if your broker is supervised, set its priority higher
; so it starts first
priority=998
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152
; ==================================&&;&&celery worker supervisor example&&; ==================================&&&; the name of your supervisord program&&[program:pichacelery]&&&; Set full path to celery program if using virtualenv&&command=/home/mosh/.virtualenvs/picha/bin/celery worker -A picha --loglevel=INFO&&&; The directory to your Django project&&directory=/home/mosh/sites/picha&&&; If supervisord is run as the root user, switch users to this UNIX user account&&; before doing any processing.&&user=mosh&&&; Supervisor will start as many instances of this program as named by numprocs&&numprocs=1&&&; Put process stdout output in this file&&stdout_logfile=/var/log/celery/picha_worker.log&&&; Put process stderr output in this file&&stderr_logfile=/var/log/celery/picha_worker.log&&&; If true, this program will start automatically when supervisord is started&&autostart=true&&&; May be one of false, unexpected, or true. If false, the process will never&&; be autorestarted. If unexpected, the process will be restart when the program&&; exits with an exit code that is not one of the exit codes associated with this&&; process’ configuration (see exitcodes). If true, the process will be&&; unconditionally restarted when it exits, without regard to its exit code.&&autorestart=true&&&; The total number of seconds which the program needs to stay running after&&; a startup to consider the start successful.&&startsecs=10&&&; Need to wait for currently executing tasks to finish at shutdown.&&; Increase this if you have very long running tasks.&&stopwaitsecs = 600&&&; When resorting to send SIGKILL to the program to terminate it&&; send SIGKILL to its whole process group instead,&&; taking care of its children as well.&&killasgroup=true&&&; if your broker is supervised, set its priority higher&&; so it starts first&&priority=998
注：确保更新这些文件的路径，以匹配你的远程服务器的文件系统。
基本上，这些supervisor 配置文件告诉supervisord如何运行并管理我们的’programs’（因为它们是由supervisord调用）。
在上面的例子中，我们已经创建了两个名为“pichacelery”和“pichacelerybeat”的supervisord程序。
现在，只需将这些文件拷贝到远程服务器的/etc/supervisor/conf.d/目录下。
我们还需要在远程服务器上创建上面脚本中提到的日志文件：
$ touch /var/log/celery/picha_worker.log
$ touch /var/log/celery/picha_beat.log
$ touch /var/log/celery/picha_worker.log $ touch /var/log/celery/picha_beat.log
最后，运行以下命令，使 Supervisor 知道它所管理的程序的存在 – 例如，pichacelery和pichacelerybeat：
$ sudo supervisorctl reread
$ sudo supervisorctl update
$ sudo supervisorctl reread $ sudo supervisorctl update
运行以下命令停止，启动，和/或检查pichacelery程序的状态：
$ sudo supervisorctl stop pichacelery
$ sudo supervisorctl start pichacelery
$ sudo supervisorctl status pichacelery
$ sudo supervisorctl stop pichacelery $ sudo supervisorctl start pichacelery $ sudo supervisorctl status pichacelery
你可以通过阅读官方文档获取Supervisor的更多信息。
7、最后提示
1. 千万不要传递Django模型对象到celery任务。为了避免模型对象在传递给celery任务之前已经改变了，传递celery的主键给celery。然后，在运行之前使用主键从数据库中获取对象。
2. 默认celery调度会在本地创建一些文件存储它的调度表。这些文件是“celerybeat-schedule.db”和“celerybeat.pid”。如果你在使用版本控制系统，比如Git（你应该使用！），请忽略这个文件，不要将它们添加到你的代码库中，因为它们是为本地运行的进程服务的。
以上就是将celery集成到一个django项目的基本介绍。
想要更多？
1. 深入研究官方celery用户指南，以了解更多信息。
2. 创建一个Fabfile来设置Supervisor和配置文件。确保添加命令到reread和 update Supervisor。
3. 从repo中获取这个项目，并打开一个Pull 请求来添加一个新的celery任务。
编码快乐！
可能感兴趣的话题
题目有错别字,"对列"应该是"队列"
o 215 回复
关于 Python 频道
Python频道分享 Python 开发技术、相关的行业动态。
新浪微博：
推荐微信号
（加好友请注明来意）
– 好的话题、有启发的回复、值得信赖的圈子
– 分享和发现有价值的内容与观点
– 为IT单身男女服务的征婚传播平台
– 优秀的工具资源导航
– 翻译传播优秀的外文文章
– 国内外的精选文章
– UI,网页，交互和用户体验
– 专注iOS技术分享
– 专注Android技术分享
– JavaScript, HTML5, CSS
– 专注Java技术分享
– 专注Python技术分享
& 2017 伯乐在线}