Celery 异步任务、定时任务
celery介绍
Celery是一个功能完备即插即用的任务队列。它使得我们不需要考虑复杂的问题,使用非常简单。celery看起来似乎很庞大,本章节我们先对其进行简单的了解,然后再去学习其他一些高级特性。 celery适用异步处理问题,当发送邮件、或者文件上传, 图像处理等等一些比较耗时的操作,我们可将其异步执行,这样用户不需要等待很久,提高用户体验。 celery的特点是:
- 简单,易于使用和维护,有丰富的文档。
- 高效,单个celery进程每分钟可以处理数百万个任务。
- 灵活,celery中几乎每个部分都可以自定义扩展。
celery非常易于集成到一些web开发框架中.
Task Queue
任务队列是一种跨线程、跨机器工作的一种机制.
任务队列中包含称作任务的工作单元。有专门的工作进程持续不断的监视任务队列,并从中获得新的任务并处理.
celery通过消息进行通信,通常使用一个叫Broker(中间人)来协client(任务的发出者)和worker(任务的处理者). clients发出消息到队列中,broker将队列中的信息派发给worker来处理。
一个celery系统可以包含很多的worker和broker,可增强横向扩展性和高可用性能。
Celery 安装
我们可以使用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
Broker
Celery需要一种解决消息的发送和接受的方式,我们把这种用来存储消息的的中间装置叫做message broker, 也可叫做消息中间人。 作为中间人,我们有几种方案可选择:
1.RabbitMQ
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 去查找自己所需要的版本软件。
2.Redis
Redis也是一款功能完备的broker可选项,但是其更可能因意外中断或者电源故障导致数据丢失的情况。 关于是有那个Redis作为Broker,可访下面网址: http://docs.celeryproject.org/en/latest/getting-started/brokers/redis.html#broker-redis
Application
使用celery第一件要做的最为重要的事情是需要先创建一个Celery实例,我们一般叫做celery应用,或者更简单直接叫做一个app。app应用是我们使用celery所有功能的入口,比如创建任务,管理任务等,在使用celery的时候,app必须能够被其他的模块导入。
1.创建应用
我们首先创建tasks.py模块, 其内容为:
from celery import Celery
# 我们这里案例使用redis作为broker
app = Celery('demo', broker='redis://:[email protected]/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
显示效果如下:
2.调用任务
任务加入到broker队列中,以便刚才我们创建的celery workder服务器能够从队列中取出任务并执行。如何将任务函数加入到队列中,可使用delay()。
进入python终端, 执行如下代码:
from tasks import my_task
my_task.delay()
执行效果如下:
我们通过worker的控制台,可以看到我们的任务被worker处理。调用一个任务函数,将会返回一个AsyncResult对象,这个对象可以用来检查任务的状态或者获得任务的返回值。
3.存储结果
如果我们想跟踪任务的状态,Celery需要将结果保存到某个地方。有几种保存的方案可选:SQLAlchemy、Django ORM、Memcached、 Redis、RPC (RabbitMQ/AMQP)。
例子我们仍然使用Redis作为存储结果的方案,任务结果存储配置我们通过Celery的backend参数来设定。我们将tasks模块修改如下:
from celery import Celery
# 我们这里案例使用redis作为broker
app = Celery('demo',
backend='redis://:[email protected]:6379/2',
broker='redis://:[email protected]: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中设置,或者通过专有的配置模块来配置。
1.直接通过app来配置
from celery import Celery
app = Celery('demo')
# 增加配置
app.conf.update(
result_backend='redis://:[email protected]:6379/2',
broker_url='redis://:[email protected]:6379/1',
)
2.专有配置文件
对于比较大的项目,我们建议配置信息作为一个单独的模块。我们可以通过调用app的函数来告诉Celery使用我们的配置模块。
配置模块的名字我们取名为celeryconfig, 这个名字不是固定的,我们可以任意取名,建议这么做。我们必须保证配置模块能够被导入。 配置模块的名字我们取名为celeryconfig, 这个名字不是固定的,我们可以任意取名,建议这么做。我们必须保证配置模块能够被导入。
下面我们在tasks.py模块 同级目录下创建配置模块celeryconfig.py:
result_backend = 'redis://:[email protected]:6379/2'
broker_url = 'redis://:[email protected]: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
在项目中使用Celery
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://:[email protected]: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
2. 存储任务结果
此处需要用到额外包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://:[email protected]: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, 执行迁移之后,会在数据库中创建一张用来存储任务结果的表:
3. 定时任务
如果我们想某日某时执行某个任务,或者每隔一段时间执行某个任务,也可以使用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://:[email protected]: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秒中就会执行一次,如果配置了存储,那么每次任务执行的结果也会被保存到对应的数据库中。
调用tasks.py的方法时,有时会出现传入的参数无法被序列表化,解决方法:将传入的参数以字典的形式包装成一个字典,确保字典能够被json序列化,及key和value都是能被json序列化的。
在centos上使用pip或者编译安装了celery却提示找不到celery这个命令时,解决方法:
export PATH=/usr/local/python3/bin:$PATH # 我的Python3资源文件都放在 /usr/local/python3 文件夹下
echo 'export PATH=/usr/local/python3/bin:$PATH' >> /etc/profile.d/python3.sh # 避免重启丢失
如果提示没有权限 就加sudo
在windows10上使用celery,使用celery -A celery_demo worker -l info命令出错,解决方法:安装协程模块pip install eventlet, 执行命令celery -A celery_demo worker -l info -P eventlet
在windows10上使用celery时,异步任务的启动命令不能和定时任务的不能一起启动,要分开使用
celery -A celery_demo worker -l info -P eventlet
celery -A celery_demo beat -l info