Celery在Django中的使用介绍

Celery简介

celery是一个简单、灵活且可靠的，处理大量消息的分布式系统，并且提供维护这样一个系统的必须工具。

它是一个专注于实时处理的任务队列，同时也支持任务调度。

何为任务队列

任务队列：是一种在线程和机器间分发任务的机制。

celery的三大组成部分

worker

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

broker（存tasks的仓库）

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

backend （存results的仓库）

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

使用场景

• 异步任务：将耗时操作任务提交给celery去异步执行，比如发送短信、邮件，消息推送、音视频处理等。
• 定时任务：定时执行某件事情，比如每天数据统计

基本命令

# 1. 启动celery服务：

#	非windows:

#   指令：celery worker -A celery_task(celery项目文件) -l info

#   windows: 需要先下载eventlet模块，pip install eventlet

#	指令： celery worker -A celery_task -l info -P eventlet

# 2. 添加任务：手动添加，需要自定义添加任务脚本；自动添加任务，在celery.py中配置

# 3. 获取结构：手动获取，需要自定义任务脚本

celery在Django项目中的使用

celery目录结构

project

	|---celery_task

    	|---celery.py # celery连接和相关配置，且名字必须是celery.py，如果要自动添加任务，那么相							关配置也在celery.py里配置；

        |---tasks.py  # 所有任务函数

    |---add_task.py   # 手动添加任务：立即任务，延时任务，定时任务；

    |---get_result.py # 获取结果

后面两个文件可以不用添加，看需求来。

使用

celery.py

from celery import Celery

# 导入时间相关包，用法看下面

from datetime import timedelta

from celery.schedules import crontab

# 因为需要调用Django项目中的models，所有需要添加Django环境

import os,django

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'bookapi.settings.dev')

django.setup()

# 添加任务的仓库，这里使用了Redis

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

# 配置定时任务

app.conf.beat_schedule = {

    'recommend-task': {

        'task': 'celery_task.tasks.recommend_num',

        # 'schedule': timedelta(seconds=20),

        'schedule': crontab(hour=24),

        'args': ()

    },

    'monthly-task': {

        'task': 'celery_task.tasks.monthly_num',

        # 'schedule': timedelta(seconds=60),

        'schedule': crontab(day_of_month=1,hour=0),

        'args': ()

    }

}

tasks.py

from .celery import app

from bookapi.apps.user import models

@app.task

def recommend_num():

    user_list = models.User.objects.all()

    # print(user_list)   ## <QuerySet [<User: admin>, <User: 18700022899>]>

    for user in user_list:

        models.User.objects.filter(username=user.username).update(recommend_nums=3)

@app.task

def monthly_num():

    user_list = models.User.objects.all()

    # print(user_list)   ## <QuerySet [<User: admin>, <User: 18700022899>]>

    for user in user_list:

        models.User.objects.filter(username=user.username).update(monthly_nums=2)