Celery简介以及Django中使用celery

[TOC]

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

消息中间件：RabbitMQ/Redis

app = Celery('任务名'， broker=‘xxx’, bankend='vvv')

Celery执行异步任务

基本使用

包架构封装

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

celery.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）添加任务：手动添加，要自定义添加任务的脚本，右键执行脚本

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


from celery import Celery

"""
broker=存储tasks的仓库 
backend=存储results的仓库 
include=[任务文件们]
"""

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

# redis有密码时
# broker = 'redis://:123456@127.0.0.1:6379/2'
# backend = 'redis://:123456@127.0.0.1:6379/3'
app = Celery(broker=broker, backend=backend, include=['celery_task.tasks'])


# 启动celery服务的命令：
# 前提：一定要进入celery_task所属的文件夹
# celery worker -A celery_task -l info -P eventlet

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)

get_result.py

from celery_task.celery import app
from celery.result import AsyncResult

id = '4e249f2d-559a-4a3e-8b43-d498b3d6355e'  # 随机的任务结果
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('任务已经开始被执行')

延时任务

包架构封装

project
  |-- celery_task
  |	  |-- __init__.py 
  |   |-- celery.py 
  |   |-- tasks.py
  |-- add_task.py
  |-- get_result.py

celery.py

from celery import Celery

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

# 启动celery服务的命令：
# 前提：一定要进入celery_task所属的文件夹
# celery worker -A celery_task -l info -P eventlet

tasks.py

from .celery import app

@app.task
def add(n, m):
    res = n + m
    print(res)
    return res

@app.task
def low(n, m):
    res = n - m
    print(res)
    return res

add_task.py

from celery_task.tasks import add, low

# 添加延迟任务
from datetime import datetime, timedelta

def eta_second(second):
    ctime = datetime.now()
    utc_ctime = datetime.utcfromtimestamp(ctime.timestamp())
    time_delay = timedelta(seconds=second)
    return utc_ctime + time_delay

# 手动添加 延迟任务
if __name__ == '__main__':
    result = add.apply_async(args=(200, 50), eta=eta_second(10))
    print(result.id)
    result = low.apply_async(args=(200, 50), eta=eta_second(8))
    print(result.id)

get_result.py

from celery_task.celery import app
from celery.result import AsyncResult

id = '74b4f1b2-36ee-4eab-afa7-3da40b7e794e'

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('任务已经开始被执行')

定时任务

包架构封装

project
  |-- celery_task
  |	  |-- __init__.py
  |   |-- celery.py
  |   |-- tasks.py

celery.py

from celery import Celery

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

# 启动celery服务的命令：
# 前提：一定要进入celery_task所属的文件夹
# celery worker -A celery_task -l info -P eventlet


# 自动添加任务
# 时区
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),
        # 'schedule': crontab(hour=8, day_of_week=1),  # 每周一早八点
        'args': (300, 150),
    },
    'my-add-task': {
        'task': 'celery_task.tasks.add',
        'schedule': timedelta(seconds=6),
        'args': (300, 150),
    }
}

# 启动 添加任务 服务的命令
# celery beat -A celery_task -l info

tasks.py

from .celery import app


@app.task
def add(n, m):
    res = n + m
    print(res)
    return res

@app.task
def low(n, m):
    res = n - m
    print(res)
    return res

异步处理Django任务

包架构封装

project
  |-- celery_task
  |	  |-- __init__.py
  |   |-- celery.py
  |   |-- tasks.py

celery.py

# 启动django依赖

import sys
sys.path.append(r'C:UsersoldboyDesktopluffyluffyapi')

import os, django
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'luffyapi.settings.dev')
django.setup()


from celery import Celery

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

# 启动celery服务的命令：
# 前提：一定要进入celery_task所属的文件夹
# celery worker -A celery_task -l info -P eventlet


# 自动添加任务
# 时区
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False

# 任务的定时配置
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {
    'django-task': {
        'task': 'celery_task.tasks.django_task',
        'schedule': timedelta(seconds=3),
        # 'schedule': crontab(hour=8, day_of_week=1),  # 每周一早八点
        'args': (300, 150),
    },
}

# 启动 添加任务 服务的命令
# celery beat -A celery_task -l info

tasks.py

from .celery import app

@app.task
def django_task():
    pass

案例：

celery.py

# 启动django依赖

# 将celery服务框架放在项目根目录下
# import sys
# sys.path.append(r'C:UsersoldboyDesktopluffyluffyapi')

import os, django
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'luffyapi.settings')
django.setup()


from celery import Celery

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

app.conf.timezone = 'Asia/Shanghai'

app.conf.enable_utc = False

# 任务的定时配置
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {
    'django-task': {
        'task': 'celery_task.tasks.django_task',
        'schedule': timedelta(seconds=10),
        # 'schedule': crontab(hour=8, day_of_week=1),  # 每周一早八点
        'args': (),
    },
}

tasks.py

from .celery import app
from django.core.cache import cache
from apps.home import models, serializers


@app.task
def django_task():
    banner_query = models.Banner.objects.filter(is_show=True, is_delete=False).order_by('-orders')
    banner_list_data = serializers.BannerModelSerializer(banner_query, many=True).data
    # 建立接口缓存
    cache.set('api_banner_list_data', banner_list_data)
    return '轮播图缓存更新完毕'