咱们在作网站后端程序开发时,会碰到这样的需求:用户须要在咱们的网站填写注册信息,咱们发给用户一封注册激活邮件到用户邮箱,若是因为各类缘由,这封邮件发送所需时间较长,那么客户端将会等待好久,形成很差的用户体验.html
那么怎么解决这样的问题呢?python
咱们将耗时任务放到后台异步执行。不会影响用户其余操做。除了注册功能,例如上传,图形处理等等耗时的任务,均可以按照这种思路来解决。 如何实现异步执行任务呢?web
咱们可以使用celery. celery除了刚才所涉及到的异步执行任务以外,还能够实现定时处理某些任务。redis
Celery是一个功能完备即插即用的任务队列。它使得咱们不须要考虑复杂的问题,使用很是简单。celery看起来彷佛很庞大,本章节咱们先对其进行简单的了解,而后再去学习其余一些高级特性。
celery适用异步处理问题,当发送邮件、或者文件上传, 图像处理等等一些比较耗时的操做,咱们可将其异步执行,这样用户不须要等待好久,提升用户体验。 celery的特色是:
简单,易于使用和维护,有丰富的文档。
高效,单个celery进程每分钟能够处理数百万个任务。
灵活,celery中几乎每一个部分均可以自定义扩展。
celery很是易于集成到一些web开发框架中.
任务队列是一种跨线程、跨机器工做的一种机制.sql
任务队列中包含称做任务的工做单元。有专门的工做进程持续不断的监视任务队列,并从中得到新的任务并处理.数据库
celery经过消息进行通讯,一般使用一个叫Broker(中间人)来协client(任务的发出者)和worker(任务的处理者). clients发出消息到队列中,broker将队列中的信息派发给worker来处理。django
一个celery系统能够包含不少的worker和broker,可加强横向扩展性和高可用性能。后端
咱们可使用python的包管理器pip来安装: pip install -U Celery 也可从官方直接下载安装包:https://pypi.python.org/pypi/celery/ tar xvfz celery-0.0.0.tar.gz cd celery-0.0.0 python setup.py build python setup.py install
Celery须要一种解决消息的发送和接受的方式,咱们把这种用来存储消息的的中间装置叫作message broker, 也可叫作消息中间人。 做为中间人,咱们有几种方案可选择:浏览器
RabbitMQ是一个功能完备,稳定的而且易于安装的broker. 它是生产环境中最优的选择。使用RabbitMQ的细节参照如下连接: http://docs.celeryproject.org/en/latest/getting-started/brokers/rabbitmq.html#broker-rabbitmq服务器
若是咱们使用的是Ubuntu或者Debian发行版的Linux,能够直接经过下面的命令安装RabbitMQ: sudo apt-get install rabbitmq-server 安装完毕以后,RabbitMQ-server服务器就已经在后台运行。若是您用的并非Ubuntu或Debian, 能够在如下网址: http://www.rabbitmq.com/download.html 去查找本身所须要的版本软件。
Redis也是一款功能完备的broker可选项,可是其更可能因意外中断或者电源故障致使数据丢失的状况。 关因而有那个Redis做为Broker,可访下面网址: http://docs.celeryproject.org/en/latest/getting-started/brokers/redis.html#broker-redis
使用celery第一件要作的最为重要的事情是须要先建立一个Celery实例,咱们通常叫作celery应用,或者更简单直接叫作一个app。app应用是咱们使用celery全部功能的入口,好比建立任务,管理任务等,在使用celery的时候,app必须可以被其余的模块导入。
咱们首先建立tasks.py模块, 其内容为:
from celery import Celery # 咱们这里案例使用redis做为broker app = Celery('demo', broker='redis://:332572@127.0.0.1/1') # 建立任务函数 @app.task def my_task(): print("任务函数正在执行....")
Celery第一个参数是给其设定一个名字, 第二参数咱们设定一个中间人broker, 在这里咱们使用Redis做为中间人。my_task函数是咱们编写的一个任务函数, 经过加上装饰器app.task, 将其注册到broker的队列中。
如今咱们在建立一个worker, 等待处理队列中的任务.打开终端,cd到tasks.py同级目录中,执行命令:
celery -A tasks worker --loglevel=info
显示效果以下:
任务加入到broker队列中,以便刚才咱们建立的celery workder服务器可以从队列中取出任务并执行。如何将任务函数加入到队列中,可以使用delay()。
进入python终端, 执行以下代码:
from tasks import my_task my_task.delay()
执行效果以下:
咱们经过worker的控制台,能够看到咱们的任务被worker处理。调用一个任务函数,将会返回一个AsyncResult对象,这个对象能够用来检查任务的状态或者得到任务的返回值。
若是咱们想跟踪任务的状态,Celery须要将结果保存到某个地方。有几种保存的方案可选:SQLAlchemy、Django ORM、Memcached、 Redis、RPC (RabbitMQ/AMQP)。
例子咱们仍然使用Redis做为存储结果的方案,任务结果存储配置咱们经过Celery的backend参数来设定。咱们将tasks模块修改以下:
from celery import Celery # 咱们这里案例使用redis做为broker app = Celery('demo', backend='redis://:332572@127.0.0.1:6379/2', broker='redis://:332572@127.0.0.1:6379/1') # 建立任务函数 @app.task def my_task(a, b): print("任务函数正在执行....") return a + b
咱们给Celery增长了backend参数,指定redis做为结果存储,并将任务函数修改成两个参数,而且有返回值。
更多关于result对象信息,请参阅下列网址:http://docs.celeryproject.org/en/latest/reference/celery.result.html#module-celery.result
Celery使用简单,配置也很是简单。Celery有不少配置选项可以使得celery可以符合咱们的须要,可是默认的几项配置已经足够应付大多数应用场景了。
配置信息能够直接在app中设置,或者经过专有的配置模块来配置。
from celery import Celery app = Celery('demo') # 增长配置 app.conf.update( result_backend='redis://:332572@127.0.0.1:6379/2', broker_url='redis://:332572@127.0.0.1:6379/1', )
对于比较大的项目,咱们建议配置信息做为一个单独的模块。咱们能够经过调用app的函数来告诉Celery使用咱们的配置模块。
配置模块的名字咱们取名为celeryconfig, 这个名字不是固定的,咱们能够任意取名,建议这么作。咱们必须保证配置模块可以被导入。 配置模块的名字咱们取名为celeryconfig, 这个名字不是固定的,咱们能够任意取名,建议这么作。咱们必须保证配置模块可以被导入。
下面咱们在tasks.py模块 同级目录下建立配置模块celeryconfig.py:
result_backend = 'redis://:332572@127.0.0.1:6379/2' broker_url = 'redis://:332572@127.0.0.1:6379/1'
tasks.py文件修改成:
from celery import Celery import celeryconfig # 咱们这里案例使用redis做为broker app = Celery('demo') # 从单独的配置模块中加载配置 app.config_from_object('celeryconfig')
更多配置: http://docs.celeryproject.org/en/latest/userguide/configuration.html#configuration
个人项目目录: TestCelery/ ├── proj │ ├── celeryconfig.py │ ├── celery.py │ ├── init.py │ └── tasks.py └── test.py celery.py内容以下: from celery import Celery # 建立celery实例 app = Celery('demo') app.config_from_object('proj.celeryconfig') # 自动搜索任务 app.autodiscover_tasks(['proj']) celeryconfig.p模块内容以下: from kombu import Exchange, Queue BROKER_URL = 'redis://:332572@127.0.0.1:6379/1' CELERY_RESULT_BACKEND = 'redis://:332572@127.0.0.1:6379/2' tasks.py模块内容以下: from proj.celery import app as celery_app # 建立任务函数 @celery_app.task def my_task1(): print("任务函数(my_task1)正在执行....") @celery_app.task def my_task2(): print("任务函数(my_task2)正在执行....") @celery_app.task def my_task3(): print("任务函数(my_task3)正在执行....") 启动worker: celery -A proj worker -l info
键入ctrl+c可关闭worker.
调用任务,可以使用delay()方法: my_task.delay(2, 2) 也可使用apply_async()方法,该方法可以让咱们设置一些任务执行的参数,例如,任务多久以后才执行,任务被发送到那个队列中等等. my_task.apply_async((2, 2), queue='my_queue', countdown=10) 任务my_task将会被发送到my_queue队列中,而且在发送10秒以后执行。 若是咱们直接执行任务函数,将会直接执行此函数在当前进程中,并不会向broker发送任何消息。 不管是delay()仍是apply_async()方式都会返回AsyncResult对象,方便跟踪任务执行状态,但须要咱们配置result_backend. 每个被吊用的任务都会被分配一个ID,咱们叫Task ID.
咱们到目前为止只是知道如何使用delay()方法,固然这个方法也是很是经常使用的。可是有时咱们并不想简单的将任务发送到队列中,咱们想将一个任务函数(由参数和执行选项组成)做为一个参数传递给另一个函数中,为了实现此目标,Celery使用一种叫作signatures的东西。
一个signature包装了一个参数和执行选项的单个任务调用。咱们可将这个signature传递给函数。
咱们先看下tasks.py模块中定义的任务函数:
from proj.celery import app as celery_app # 建立任务函数 @celery_app.task def my_task1(): print("任务函数(my_task1)正在执行....") @celery_app.task def my_task2(): print("任务函数(my_task2)正在执行....") @celery_app.task def my_task3(): print("任务函数(my_task3)正在执行....")
咱们将my_task1()任务包装称一个signature:
t1 = my_task1.signatures(countdown=10) t1.delay()
这些primitives自己就是signature对象,所以它们能够以多种方式组合成复杂的工做流程。primitives以下:
group: 一组任务并行执行,返回一组返回值,并能够按顺序检索返回值。
chain: 任务一个一个执行,一个执行完将执行return结果传递给下一个任务函数.
tasks.py模块以下:
from proj.celery import app as celery_app # 建立任务函数 @celery_app.task def my_task1(a, b): print("任务函数(my_task1)正在执行....") return a + b @celery_app.task def my_task2(a, b): print("任务函数(my_task2)正在执行....") return a + b @celery_app.task def my_task3(a, b): print("任务函数(my_task3)正在执行....") return a + b
group案例以下(test.py模块):
from proj.tasks import my_task1 from proj.tasks import my_task2 from proj.tasks import my_task3 from celery import group # 将多个signature放入同一组中 my_group = group((my_task1.s(10, 10), my_task2.s(20, 20), my_task3.s(30, 30))) ret = my_group() # 执行组任务 print(ret.get()) # 输出每一个任务结果

chain案例以下(test.py模块):
from proj.tasks import my_task1 from proj.tasks import my_task2 from proj.tasks import my_task3 from celery import chain # 将多个signature组成一个任务链 # my_task1的运行结果将会传递给my_task2 # my_task2的运行结果会传递给my_task3 my_chain = chain(my_task1.s(10, 10) | my_task2.s(20) | my_task3.s(30)) ret = my_chain() # 执行任务链 print(ret.get()) # 输出最终结果
假如咱们有两个worker,一个worker专门用来处理邮件发送任务和图像处理任务,一个worker专门用来处理文件上传任务。 咱们建立两个队列,一个专门用于存储邮件任务队列和图像处理,一个用来存储文件上传任务队列。 Celery支持AMQP(Advanced Message Queue)全部的路由功能,咱们也可使用简单的路由设置将指定的任务发送到指定的队列中. 咱们须要配置在celeryconfig.py模块中配置 CELERY_ROUTES 项, tasks.py模块修改以下: from proj.celery import app as celery_app @celery_app.task def my_task1(a, b): print("my_task1任务正在执行....") return a + b @celery_app.task def my_task2(a, b): print("my_task2任务正在执行....") return a + b @celery_app.task def my_task3(a, b): print("my_task3任务正在执行....") return a + b @celery_app.task def my_task4(a, b): print("my_task3任务正在执行....") return a + b @celery_app.task def my_task5(): print("my_task5任务正在执行....") @celery_app.task def my_task6(): print("my_task6任务正在执行....") @celery_app.task def my_task7(): print("my_task7任务正在执行....") 咱们经过配置,将send_email和upload_file任务发送到queue1队列中,将image_process发送到queue2队列中。 咱们修改celeryconfig.py: broker_url='redis://:@127.0.0.1:6379/1' result_backend='redis://:@127.0.0.1:6379/2' task_routes=({ 'proj.tasks.my_task5': {'queue': 'queue1'}, 'proj.tasks.my_task6': {'queue': 'queue1'}, 'proj.tasks.my_task7': {'queue': 'queue2'}, }, ) test.py: from proj.tasks import * # 发送任务到路由指定的队列中 my_task5.delay() my_task6.delay() my_task7.delay()   开启两个worker服务器,分别处理两个队列: ```python celery -A proj worker --loglevel=info -Q queue1 celery -A proj worker --loglevel=info -Q queue2 咱们一样也能够经过apply_aynsc()方法来设置任务发送到那个队列中: my_task1.apply_async(queue='queue1') 咱们也可设置一个worker服务器处理两个队列中的任务: celery -A proj worker --loglevel=info -Q queue1,queue2
celery beat是一个调度器,它能够周期内指定某个worker来执行某个任务。若是咱们想周期执行某个任务须要增长beat_schedule配置信息. broker_url='redis://:@127.0.0.1:6379/1' result_backend='redis://:@127.0.0.1:6379/2' # 指定任务发到那个队列中 task_routes=({ 'proj.tasks.my_task5': {'queue': 'queue1'}, 'proj.tasks.my_task6': {'queue': 'queue1'}, 'proj.tasks.my_task7': {'queue': 'queue2'}, }, ) # 配置周期性任务, 或者定时任务 beat_schedule = { 'every-5-seconds': { 'task': 'proj.tasks.my_task8', 'schedule': 5.0, # 'args': (16, 16), } } tasks.py模块内容以下: from proj.celery import app as celery_app @celery_app.task def my_task1(a, b): print("my_task1任务正在执行....") return a + b @celery_app.task def my_task2(a, b): print("my_task2任务正在执行....") return a + b @celery_app.task def my_task3(a, b): print("my_task3任务正在执行....") return a + b @celery_app.task def my_task4(a, b): print("my_task3任务正在执行....") return a + b @celery_app.task def my_task5(): print("my_task5任务正在执行....") @celery_app.task def my_task6(): print("my_task6任务正在执行....") @celery_app.task def my_task7(): print("my_task7任务正在执行....") # 周期执行任务 @celery_app.task def my_task8(): print("my_task8任务正在执行....") 启动woker处理周期性任务: celery -A proj worker --loglevel=info --beat 若是咱们想指定在某天某时某分某秒执行某个任务,能够执行cron任务, 增长配置信息以下: beat_schedule = { 'every-5-minute': { 'task': 'proj.tasks.period_task', 'schedule': 5.0, 'args': (16, 16), }, 'add-every-monday-morning': { 'task': 'proj.tasks.period_task', 'schedule': crontab(hour=7, minute=30, day_of_week=1), 'args': (16, 16), }, } crontab例子: http://docs.celeryproject.org/en/latest/userguide/periodic-tasks.html 开启一个celery beat服务: celery -A proj beat celery须要保存上次任务运行的时间在数据文件中,文件在当前目录下名字叫celerybeat-schedule. beat须要访问此文件: celery -A proj beat -s /home/celery/var/run/celerybeat-schedule
1. 配置celery 建立django项目celery_demo, 建立应用demo: django-admin startproject celery_demo python manage.py startapp demo
在celery_demo模块中建立celery.py模块, 文件目录为:

celery.py模块内容为: from celery import Celery from django.conf import settings import os # 为celery设置环境变量 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'celery_demo.settings') # 建立应用 app = Celery("demo") # 配置应用 app.conf.update( # 配置broker, 这里咱们用redis做为broker BROKER_URL='redis://:332572@127.0.0.1:6379/1', ) # 设置app自动加载任务 # 从已经安装的app中查找任务 app.autodiscover_tasks(settings.INSTALLED_APPS) 在应用demo引用建立tasks.py模块, 文件目录为:
咱们在文件内建立一个任务函数my_task: from celery_demo.celery import app import time # 加上app对象的task装饰器 # 此函数为任务函数 @app.task def my_task(): print("任务开始执行....") time.sleep(5) print("任务执行结束....") 在views.py模块中建立视图index: from django.shortcuts import render from django.http import HttpResponse from .tasks import my_task def index(request): # 将my_task任务加入到celery队列中 # 若是my_task函数有参数,可经过delay()传递 # 例如 my_task(a, b), my_task.delay(10, 20) my_task.delay() return HttpResponse("<h1>服务器返回响应内容!</h1>") 在celey_demo/settings.py配置视图路由: from django.conf.urls import url from django.contrib import admin from demo.views import index urlpatterns = [ url(r'^admin/', admin.site.urls), url(r'^$', index), ] 建立worker等待处理celery队列中任务, 在终端执行命令: celery -A celery_demo worker -l info
启动django测试服务器:
python manage.py runserver

此处须要用到额外包django_celery_results, 先安装包: pip install django-celery-results 在celery_demo/settings.py中安装此应用: INSTALLED_APPS = [ 'django.contrib.admin', 'django.contrib.auth', 'django.contrib.contenttypes', 'django.contrib.sessions', 'django.contrib.messages', 'django.contrib.staticfiles', 'demo', 'django_celery_results', # 注意此处应用名为下划线 ] 回到celery_demo/celery.py模块中,增长配置信息以下: from celery import Celery from django.conf import settings import os # 为celery设置环境变量 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'celery_demo.settings') # 建立应用 app = Celery("demo") # 配置应用 app.conf.update( # 配置broker, 这里咱们用redis做为broker BROKER_URL='redis://:332572@127.0.0.1:6379/1', # 使用项目数据库存储任务执行结果 CELERY_RESULT_BACKEND='django-db', ) # 设置app自动加载任务 # 从已经安装的app中查找任务 app.autodiscover_tasks(settings.INSTALLED_APPS) 建立django_celery_results应用所需数据库表, 执行迁移文件: python manage.py migrate django_celery_results 我这里使用的是django默认的数据库sqlit, 执行迁移以后,会在数据库中建立一张用来存储任务结果的表:
再次从浏览器发送请求, 任务执行结束以后,将任务结果保存在数据库中:

若是咱们想某日某时执行某个任务,或者每隔一段时间执行某个任务,也可使用celery来完成. 使用定时任务,须要安装额外包: pip install django_celery_beat 首先在settings.py中安装此应用: INSTALLED_APPS = [ 'django.contrib.admin', 'django.contrib.auth', 'django.contrib.contenttypes', 'django.contrib.sessions', 'django.contrib.messages', 'django.contrib.staticfiles', 'demo', 'django_celery_results', 'django_celery_beat', # 安装应用 ] 在celery_demo/celery.py模块中增长定时任务配置: from celery import Celery from django.conf import settings import os # 为celery设置环境变量 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'celery_demo.settings') # 建立应用 app = Celery("demo") # 配置应用 app.conf.update( # 配置broker, 这里咱们用redis做为broker BROKER_URL='redis://:332572@127.0.0.1:6379/1', # 使用项目数据库存储任务执行结果 CELERY_RESULT_BACKEND='django-db', # 配置定时器模块,定时器信息存储在数据库中 CELERYBEAT_SCHEDULER='django_celery_beat.schedulers.DatabaseScheduler', ) # 设置app自动加载任务 # 从已经安装的app中查找任务 app.autodiscover_tasks(settings.INSTALLED_APPS) 因为定时器信息存储在数据库中,咱们须要先生成对应表, 对diango_celery_beat执行迁移操做,建立对应表: python manage.py migrate django_celery_beat
咱们可登陆网站后台Admin去建立对应任务, 首先咱们先在tasks.py模块中增长新的任务,用于定时去执行(5秒执行一次) from celery_demo.celery import app import time # 用于定时执行的任务 @app.task def interval_task(): print("我每隔5秒钟时间执行一次....") 首先建立后台管理员账号: python manage.py createsuperuser 登陆管理后台Admin:
其中Crontabs用于定时某个具体时间执行某个任务的时间,Intervals用于每隔多久执行任务的事件,具体任务的执行在Periodic tasks表中建立。
咱们要建立每隔5秒执行某个任务,因此在Intervals表名后面点击Add按钮:

而后在Periodic tasks表名后面,点击Add按钮,添加任务:

启动定时任务: celery -A celery_demo worker -l info --beat
任务每隔5秒中就会执行一次,若是配置了存储,那么每次任务执行的结果也会被保存到对应的数据库中。