Celery在Django中的简单应用

- 1、celery 架构
- 2、celery 的简单使用

1、celery 架构

Celery是一个简单、灵活且可靠的，处理大量消息的分布式系统，专注于实时处理的异步任务队列。同时也支持任务调度。需要注意的是，celery并不支持Windows，所以Windows相关的问题很可能在官方无法得到回应。

Celery的架构由三部分组成，消息中间件（message broker）、任务执行单元（worker）和任务执行结果存储（task result store）组成。

1）可以不依赖任何服务器，通过自身命令，启动服务(内部支持socket)
2）celery服务为为其他项目服务提供异步解决任务需求的
注：会有两个服务同时运行，一个是项目服务，一个是celery服务，项目服务将需要异步处理的任务交给celery服务，celery就会在需要时异步完成项目的需求。

使用场景：

异步执行：解决耗时任务,将耗时操作任务提交给Celery去异步执行，比如发送短信/邮件、消息推送、音视频处理等等。

延迟执行：解决延迟任务。

定时执行：解决周期(周期)任务,比如每天数据统计。

2、celery 的简单使用

首先需要安装celery模块：pip install celery

2.1 最简单的使用

写一个py文件（celery_task.py）：

from celery import Celery

broker='redis://127.0.0.1:6379/1'  #broker任务队列
backend='redis://127.0.0.1:6379/2'  # 结构存储，执行完的结果存在这
app=Celery(__name__,broker=broker,backend=backend)

#添加任务逻辑
@app.task
def add(x,y):
    print(x,y)
    return x+y

启动worker

# cmd或者Terminal中用命令来执行
# 非windows
celery worker -A celery_task -l info

# windows：
pip install eventlet
celery worker -A celery_task -l info -P eventlet

发送任务

from celery_task import add

ret=add.delay(5,4)  #向broker中添加一个任务
print(ret)   # ret是celery返回的任务id号，可以使用该ID号取回任务处理的结果

查看任务执行结果

from celery_task import app
from celery.result import AsyncResult

id = '3e397fd7-e0c1-4c5c-999c-2655a96793bb' # 此处的ID即为发送任务时celery返回的任务id。

if __name__ == '__main__':
	async = AsyncResult(id=id, app=app)
    '''
    在Python3.7及以后的版本中async成为了保留关键字，建议使用3.7以下版本或者更换该变量名。
    '''
    if async.successful():
        result = async.get()
        print(result)
    elif async.failed():
        print('任务失败')
    elif async.status == 'PENDING':
        print('任务等待中被执行')
    elif async.status == 'RETRY':
        print('任务异常后正在重试')
    elif async.status == 'STARTED':
        print('任务已经开始被执行')

2.2 包结构的使用

新建一个包（celery_task）

# 包结构
-celery_task
    -__init__.py
    -celery.py  # 该py文件一点要叫celery.py
    -tasks.py

celery.py

from celery import Celery

broker='redis://127.0.0.1:6379/1'  
backend='redis://127.0.0.1:6379/2' 

app=Celery(__name__,broker=broker,backend=backend,include=['celery_task.tasks',])

tasks.py

from .celery import app
@app.task
def add(x,y):
    print(x,y)
    return x+y

@app.task
def mutile(x,y):
    print(x,y)
    return x*y

启动worker
发送任务

from celery_task.tasks import add,mutile
# 提交异步任务
ret=add.delay(6,7)
print(ret) 

# 提交延迟任务
from datetime import datetime, timedelta
# 需要utc时间
eta=datetime.utcnow() + timedelta(seconds=10) 
ret=mutile.apply_async(args=(240, 50), eta=eta)  # 延迟10s执行mutile任务，args为其传值。
print(ret)

根据返回的id取处理结果

2.3 celery执行定时任务

在celery_task 包的celery.py 中：

from celery import Celery
from datetime import timedelta
from celery.schedules import crontab

broker='redis://127.0.0.1:6379/1' 
backend='redis://127.0.0.1:6379/2' 

app=Celery(__name__,broker=broker,backend=backend,include=['celery_task.tasks',])

# 执行定时任务
# 时区
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC 
app.conf.enable_utc = False  #  不使用utc时间

# 任务的定时配置
app.conf.beat_schedule = {
    'add-task': {
        'task': 'celery_task.tasks.add',
        # 'schedule': timedelta(seconds=3),  # 每隔3秒
        'schedule': crontab(hour=8, day_of_week=1),  # 每周一早八点
        'args': (300, 150), # 传参
    }
}

除了需要启动worker，由于是定时自动提交，还需要启动beat。

-celery worker -A celery_task -l info -P eventlet
-celery beat -A celery_task -l info

2.4 celery 在Django中的使用

在Django项目的根目录新建一个celery_task包。其中celery.py示例：

from celery import Celery

# 加载django环境,否则无法使用Django的model或其他内容
import os
import django
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "luffyapi.settings.dev") # "luffyapi.settings.dev"为项目配置文件，需要注意路径
django.setup()

broker='redis://127.0.0.1:6379/1'  
backend='redis://127.0.0.1:6379/2' 

app=Celery(__name__,broker=broker,backend=backend,include=['celery_task.tasks',])


# 执行定时任务
# 时区
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False

# 任务的定时配置
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {
    'add-task': {
        'task': 'celery_task.tasks.update',
        'schedule': timedelta(seconds=30), # 定时30秒执行刷新任务，将数据库中的数据缓存到Redis中
    }
}

tasks.py：

from .celery import app


@app.task
def update():
    from home import serializer
    from home import models
    from django.conf import settings
    from django.core.cache import cache
    
    queryset_banner = models.Banner.objects.filter(is_delete=False,is_show=True).order_by('display_order')[
       :settings.BANNER_COUNTER]
   serializer_banner=serializer.BannerModelSerilaizer(instance=queryset_banner,many=True)
    cache.set('banner_list',serializer_banner.data)
    
    return True

其中celery在Django项目中的使用最重要的就是加载Django的环境，以便于celery在处理任务的时候可以使用Django的model及其他内容。其中tasks只要遵循官网的目录结果，可以在Django的每个APP下建立一个tasks.py，celery可以自动识别。

"""
celery框架django项目工作流程
1）加载django配置环境
2）创建Celery框架对象app，配置broker和backend，得到的app就是worker
3）给worker对应的app添加可处理的任务函数，用include配置给worker的app
4）完成提供的任务的定时配置app.conf.beat_schedule
5）启动celery服务，运行worker，执行任务
6）启动beat服务，运行beat，添加任务

重点：由于采用了django的反射机制，使用celery.py所在的celery_task包必须放置项目的根目录下
"""

Author：Laoqi

山山仙人博客