Celery是一个功能完备即插即用的任务队列。它使得我们不需要考虑复杂的问题,使用非常简单。celery看起来似乎很庞大,本文我们先对其进行简单的了解,然后再去学习其他一些高级特性。
全套笔记直接地址: 请移步这里
共 4 章,12 子模块
介绍一下如何调用任务,队列路由.
1. signature
我们到目前为止只是学习了如何使用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()
2. Primitives
这些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服务器,分别处理两个队列:
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://:[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("服务器返回响应内容!")
在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秒中就会执行一次,如果配置了存储,那么每次任务执行的结果也会被保存到对应的数据库中。