Celery异步任务框架
Celery架构
Celery的安装配置
两种celery任务结构：提倡用包管理，结构更清晰
Celery执行异步任务
基本结构
包架构封装（多任务结构）–推荐的使用方法
包架构封装使用
高级使用:定时任务（一样使用包结构封装）
流程分析：
定时任务使用（需要在启动worker之外，再起一个beat，相当于雇佣的仆人来提交任我到broker）
示例：在luffy项目中使用celery设置定时任务，定时更新redis中的轮播图数据(放项目根目录下)

Celery
官方
Celery 官网：http://www.celeryproject.org/

Celery 官方文档英文版：http://docs.celeryproject.org/en/latest/index.html

Celery 官方文档中文版：http://docs.jinkan.org/docs/celery/

Celery是一个简单、灵活且可靠的，处理大量消息的分布式系统

专注于实时处理的异步任务队列

同时也支持任务调度

注意：

# Celery is a project with minimal funding, so we don’t support Microsoft Windows. Please don’t open any issues related to that platform.

Celery异步任务框架

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

人是一个独立运行的服务 | 医院也是一个独立运行的服务
    正常情况下，人可以完成所有健康情况的动作，不需要医院的参与；但当人生病时，就会被医院接收，解决人生病问题
    人生病的处理方案交给医院来解决，所有人不生病时，医院独立运行，人生病时，医院就来解决人生病的需求
"""

Celery架构

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

消息中间件
Celery本身不提供消息服务，但是可以方便的和第三方提供的消息中间件集成。包括，RabbitMQ, Redis等等

任务执行单元
Worker是Celery提供的任务执行的单元，worker并发的运行在分布式的系统节点中。

任务结果存储
Task result store用来存储Worker执行的任务的结果，Celery支持以不同方式存储任务的结果，包括AMQP, redis等

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

延迟执行：解决延迟任务

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

Celery的安装配置

# pip install celery==4.4.6

消息中间件：RabbitMQ/Redis

app=Celery(‘任务名’, broker=’xxx’, backend=’xxx’)

两种celery任务结构：提倡用包管理，结构更清晰


# 如果 Celery对象:Celery(...) 是放在一个模块下的
# 1）终端切换到该模块所在文件夹位置：scripts
# 2）执行启动worker的命令：celery worker -A 模块名 -l info -P eventlet
# 注：windows系统需要eventlet支持，Linux与MacOS直接执行：celery worker -A 模块名 -l info   # -l info表示日志级别为info
# 注：模块名随意


# 如果 Celery对象:Celery(...) 是放在一个包下的
# 1）必须在这个包下建一个celery.py的文件，将Celery(...)产生对象的语句放在该文件中
# 2）执行启动worker的命令：celery worker -A 包名 -l info -P eventlet
# 注：windows系统需要eventlet支持，Linux与MacOS直接执行：celery worker -A 模块名 -l info
# 注：包名随意

Celery执行异步任务

基本结构

基本结构使用

# 注，流程
# 1）创建app + 任务

# 2）启动celery(app)服务：
    # 非windows
        # 命令：celery worker -A celery_task -l info
    # windows：（需要安装eventlet模块）
        # pip install eventlet
        # celery worker -A celery_task -l info -P eventlet  # -l info 表示日志级别为info

# 3）添加任务到broker：手动添加，要自定义添加任务的脚本，右键执行脚本

# 4）获取结果：手动获取，要自定义获取任务的脚本，右键执行脚本

celery_task.py

# 创建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)


# 2 添加任务(使用这个装饰器装饰，@app.task)，app就是上面生成的app对象
@app.task
def add(x, y):
    print(x, y)
    return x + y

add_task.py

# 添加任务到broker
from celery_task import add

# add(3,4)  # 直接执行，不会被添加到broker中
ret = add.delay(5, 4)  # 向broker中添加一个任务，.delay是装饰器给加上的
print(ret.id)  # ret.id 是执行的任务id

get_result.py

# 查看任务执行结果

from celery_task import app
from celery.result import AsyncResult
id = 'b4d8a28c-d935-476d-a982-0eb55eefe195'
if __name__ == '__main__':
    async = AsyncResult(id=id, app=app)
    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('任务已经开始被执行')

启动

# cd scripts/t_celery
1)启动worker
celery worker -A celery_task -l info -P eventlet

2）运行add_task添加任务到broker

3）打印结果
从worker界面获取任务uuid，贴到get_result.py中

包架构封装（多任务结构）–推荐的使用方法

project
    ├── celery_task      # celery包
    │   ├── __init__.py # 包文件
    │   ├── celery.py   # celery连接和配置相关文件，且名字必须叫celery.py
    │   └── tasks.py    # 所有任务函数
    ├── add_task.py      # 添加任务
    └── get_result.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(broker=broker, backend=backend, include=['celery_task.tasks'])  # 需要把celery_task包下的tasks任务文件include注册一下

tasks.py

from .celery import app
import time
@app.task
def add(n, m):
    print(n)
    print(m)
    time.sleep(10)
    print('n+m的结果：%s' % (n + m))
    return n + m

@app.task
def low(n, m):
    print(n)
    print(m)
    print('n-m的结果：%s' % (n - m))
    return n - m

add_task.py

from celery_task import tasks

# 添加立即执行任务(异步)
t1 = tasks.add.delay(10, 20)
t2 = tasks.low.delay(100, 50)
print(t1.id)    # 2d4ad592-9548-4c7c-8df4-7f8583e8a1b1


# 添加延迟任务
from datetime import datetime, timedelta
# 需要utc时间
eta=datetime.utcnow() + timedelta(seconds=10)  # 延迟10秒
# eta=datetime.utcnow() + timedelta(days=1)  # 延迟1天
ret = tasks.low.apply_async(args=(200, 50), eta=eta)   # 通过args向任务函数low中传递参数
print(ret.id)    # 44bbf79c-0e34-41ed-a092-5d37e51d6cfa

get_result.py

from celery_task.celery import app

from celery.result import AsyncResult

id = '21325a40-9d32-44b5-a701-9a31cc3c74b5'
if __name__ == '__main__':
    async = AsyncResult(id=id, app=app)
    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('任务已经开始被执行')

启动

# cd scripts
# celery worker -A celery_task -l info -P eventlet
# 再运行 add_task.py

高级使用:定时任务（一样使用包结构封装）


project
    ├── celery_task      # celery包
    │   ├── __init__.py # 包文件
    │   ├── celery.py   # celery连接和配置相关文件，且名字必须叫celery.py
    │   └── tasks.py    # 所有任务函数
    ├── add_task.py      # 添加任务
    └── get_result.py   # 获取结果

流程分析：

# 1）创建app + 任务

# 2）启动celery(app)服务：
# 非windows
# 命令：celery worker -A celery_task -l info
# windows：
# pip3 install eventlet
# celery worker -A celery_task -l info -P eventlet

# 3）添加任务：自动添加任务，所以要启动一个添加任务的服务
# 命令：celery beat -A celery_task -l info

# 4）获取结果

定时任务使用（需要在启动worker之外，再起一个beat，相当于雇佣的仆人来提交任我到broker）

celery.py

from celery import Celery

broker = 'redis://127.0.0.1:6379/1'
backend = 'redis://127.0.0.1:6379/2'
app = Celery(broker=broker, backend=backend, include=['celery_task.tasks'])


# 时区，使用亚洲上海时区，和下面UTC是否启用是一对，两者只能使用一个
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False

# 任务的定时配置
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {
    'low-task': {
        'task': 'celery_task.tasks.low',
        'schedule': timedelta(seconds=3),    # 每三秒执行一次，beat就会向broker提交一次任务
        # 'schedule': crontab(hour=8, day_of_week=1),  # 每周一早八点
        'args': (300, 150),
    }
}

# 2 启动服务除了worker，还要启动beat
    # celery worker -A celery_task -l info -P eventlet
    # celery beat -A celery_task -l info

tasks.py

from .celery import app
import time
@app.task
def add(n, m):
    print(n)
    print(m)
    time.sleep(10)
    print('n+m的结果：%s' % (n + m))
    return n + m

@app.task
def low(n, m):
    print(n)
    print(m)
    print('n-m的结果：%s' % (n - m))
    return n - m

get_result.py

from celery_task.celery import app

from celery.result import AsyncResult

id = '21325a40-9d32-44b5-a701-9a31cc3c74b5'
if __name__ == '__main__':
    async = AsyncResult(id=id, app=app)
    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('任务已经开始被执行')

示例：在luffy项目中使用celery设置定时任务，定时更新redis中的轮播图数据(放项目根目录下)

celery.py

from celery import Celery
broker = 'redis://127.0.0.1:6379/1'
backend = 'redis://127.0.0.1:6379/2'
app = Celery(broker=broker, backend=backend, include=['celery_task.tasks'])  # 需要把celery_task包下的tasks任务文件include注册一下

# 时区，使用亚洲上海时区，和下面UTC是否启用是一对，两者只能使用一个
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False

# 任务的定时配置
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {
    'flush-banner-task': {
        'task': 'celery_task.tasks.flush_banner_cache',
        'schedule': timedelta(seconds=3),    # 每分钟执行一次，beat就会向broker提交一次任务
        # 'schedule': crontab(hour=8, day_of_week=1),  # 每周一早八点
        # 'args': (),
    }
}

tasks.py

# 添加任务

# 加载django环境
import os
import django
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "luffyapi.settings.dev")
django.setup()

# 下面几个导入，如果是使用绝对路径导入报错，可以将导入放在下面任务函数内部试试
from django.conf import settings
from .celery import app
from django.core.cache import cache
from home import models, serializers

@app.task
def flush_banner_cache():
    queryset = models.Banner.objects.filter(is_delete=False, is_show=True).order_by('display_order')[
               :settings.BANNER_COUNT]
    banner_list = serializers.BannerSerializer(queryset, many=True).data
    # 拿不到request对象，所以头像的连接base_url要自己组装
    for banner in banner_list:
        banner['img'] = 'http://127.0.0.1:8000%s' % banner['img']
    print(banner_list)
    cache.set('banner_list', banner_list, 60*1)
    print(cache.get('banner_list'))
    return True

终端在luffyapi执行命令

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

文档更新时间: 2022-09-19 10:30 作者：李延召

第22篇 celery的使用