Celery-分布式任务队列

1. 定义

Celery 是一个简单、灵活且可靠的分布式系统，用于处理大量消息，同时为作提供维护此类系统所需的工具。
它是一个专注于实时处理的任务队列，同时也支持任务调度

• 安装

pip install celery

2.celery执行流程

• producer-生产者
  异步任务、定时任务

• broker
  消息传输的中间件，生产者一旦有消息发送，将发至broker；【RQ，redis】

• worker - 工作者
  消费/执行broker中消息/任务的进程

• backend
  用于存储消息/任务结果，如果需要跟踪和查询任务状态，则需添加要配置相关
  

3.使用celery

1. 创建异步任务函数[tasks.py] - @app.task

import timefrom celery import Celery# 1.初始化celery对象
# 通过使用本机redis且没有密码，使用远程redis有密码格式为'redis://:密码@ip:6379/1'
app = Celery("dashopt", broker="redis://:@127.0.0.1:6379/1", backend="redis://:@127.0.0.1:6379/1")# 2.创建异步任务函数
@app.task()
def send_message(m, n):print("hahahaha")time.sleep(10)print("哈哈哈哈哈")return m + n

2. 终端启动worker

   celery -A 文件名 worker -l info

celery -A tasks worker -l info

3. 将异步任务推送到消息中间件broker中

   异步函数名.delay(参数1, 参数2)
   另外起一个终端shell窗口，执行以下代码

from tasks import send_message
send_message.delay("10", "20")

4.Django和Celery结合流程**

1. 创建django项目

django-admin startproject djcelery

2. 创建celery配置文件【和settings.py同路径】

"""celery配置文件
"""
import os
from celery import Celery
from django.conf import settings# 1.为celery设置环境变量
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "djcelery.settings")# 2.初始化celery应用
app = Celery("dashopt", broker="redis://:@127.0.0.1:6379/1")# 3.设置自动发现异步任务
app.autodiscover_tasks(settings.INSTALLED_APPS)

3. 在应用下创建tasks.py【存放异步任务】

import time
from djcelery.celery import app@app.task()
def async_task():print("111")time.sleep(5)print("222")

4. 视图函数中调用异步任务【delay(参数1, 参数2)】

import time
from django.shortcuts import render
from django.http import HttpResponse
from .tasks import async_task# Create your views here.
def test_celery(request):print("hahahahha")# 有一个阻塞任务，比如：发邮件async_task.delay()return HttpResponse(time.strftime("%H:%M:%S"))

5. 终端启动celery

 `celery -A dashopt worker -l info`

6. 浏览器中测试【注册用户】

5.使用场景

Celery 就是 Django 的「分布式任务外挂」——把耗时的、需要异步的、定时的工作丢给它，后台慢慢跑，前端瞬间返回。