celery

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

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

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

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

celery是一个分布式的异步任务框架

celery能做什么？

• 异步任务：项目中同步的操作，可以通过celery做成异步。
• 延迟任务：隔一会在执行任务。
• 定时任务；每个多长时间干什么事。

注意： 如果项目仅向左定时任务，没有必要使用celery，使用apscheduler。

为什么说celery是异步任务框架？

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

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

celery架构

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

• broker:任务中间件同等于消息队列中间件(我们这用redis)，存储任务(对应该程序中的函数)，celery本身不提供，需要借助第三方：redis,rabbitmq..
• worker: 任务执行单元，真正指向任务的进程，celery提供的，worker并发的运行在分布式(起多台机器)的系统节点中。
• backend: 结果存储，任务执行结果存在某个地方，借助于第三方: redis

celery的安装配置

安装:

pip install celery

由于我们的消息中间件和消息存储都需要用到redis,所以也需要起redis.

导入模块：

from celery import Celery

Celery执行异步执行任务

基本结构

查看Celery发现都是默认值参数，可传可不传。

创建一个celery_task.py(文件名随意)文件

from celery import Celery

# 结果存储
# backend='redis://:密码@127.0.0.1:6379/2' # 有密码的写法
backend = 'redis://127.0.0.1:6379/2'  # 无密码的写法

# 消息中间件
# broker='redis://:123456@127.0.0.1:6379/1' # 有密码写法
broker = 'redis://127.0.0.1:6379/1'  # 无密码写法

app = Celery('test', broker=broker, backend=backend)


# 写任务，使用装饰器后才会变成celery任务
@app.task
def add(a, b):
    return a + b

再新建一个celery.py文件

提交任务

同步提交：

from celery_task import add

# 提交任务

# 同步提交
res = add(7,8)
print(res)

异步提交：

# 异步提交,必须用apply_async传，使用位置传参
"此时任务就提交到redis中的消息队列中间件，或者说是任务中间件"
res = add.apply_async(args=[7,8]) 
# 任务id号：cefadaf8-5007-45c1-a3a1-08a8e2c14ea8
print(res)

补充：中间件，不是单值Django中间件，中间价的概念非常大，中间件位于某与某之间的。

• 数据库中间件：应用程序和数据库之间有一个东西
• 服务器中间件：web服务和浏览器之间有一个东西：Nginx...
• 消息队列中间件：程序与程序之间：redis，rabbitmq...

此时任务被提交到redis中，等待worker执行该任务，启动worker

启动worker执行任务---》使用命令启动

非windows

• celery -A celery_task worker -l info

windows：

• pip3 install eventlet

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

我是mac的，使用第一种，执行命令时路径必须在创建的任务路径。

查询结果：任务被celery执行完了，结果放到redis中了，查询结果，使用代码 AsyncResult

通过代码把结果取出来

from celery_task import app  # 借助于app
from celery.result import AsyncResult  # 导入一个类，来查询结果
id = '3e475ea0-8afe-4c82-b7f2-a9141edcedca]'
if __name__ == '__main__':
    res = AsyncResult(id=id, app=app)  # 根据id，去哪个app中找哪个任务，
    if res.successful(): # 执行成功
        result = res.get()
        print('任务执行成功')
        print(result) # 15
    elif res.failed():
        print('任务失败')
    elif res.status == 'PENDING':
        print('任务等待中被执行')
    elif res.status == 'RETRY':
        print('任务异常后正在重试')
    elif res.status == 'STARTED':
        print('任务已经开始被执行')

使用celery的异步秒杀场景分析

原始同步场景

100个人，秒杀3个商品--->100个人在浏览器等着开始---》一旦开始--->瞬间100个人同时发送秒杀请求到后端----》---->假设秒杀函数执行2s钟---》100个请求在2s内，一直跟后端连着，假设我的并发量是100，这两秒钟，其他任何人都访问不了了
假设 150人来秒杀---》最多能承受100个人，50个人就请求不了---》不友好

异步场景

100个人，秒杀3个商品--->100个人在浏览器等着开始---》一旦开始--->瞬间100个人同时发送秒杀请求到后端----》---->假设秒杀函数执行2s钟---》当前100个请求，过来，使用celery提交100个任务，请求立马返回--->这样的话，2s内能提交特别多的任务，可以接收特别多人发的请求---》后台使用worker慢慢的执行秒杀任务---》多起几个worker---》过了一会，所有提交的任务都执行完了

提交完任务，返回前端---》前端使用个动态图片盖住页面，显示您正在排队，每个2s钟，向后端发送一次ajax请求，带着id号，查询结果是否完成，如果没完成---》再等2s钟--->如果秒杀成功了，显示恭喜您，成了---》如果没有成功，显示很遗憾，没有秒到

celery包结构

为什么需要包结构？

可以把多个任务放在多个文件中。

-celery_task  # 包,包名可随意。
    -__init__.py
    -celery.py  # 写app的py文件，必须有，并且名字必须为celery
    -home_task.py # 任务1 
    -order_task.py # 任务2
    -user_task.py # 任务3
--------------下面这些，跟上面可能在不同项目中----------------    
add_task.py   # 提交任务，django中提交
get_result.py # 查询结果，django中查询

home_task.py

from .celery import app
# from scripts.celery_task.celery import app


@app.task
def send_sms(phone):
    print('手机号：%s，发送成功' % phone)
    return True

order_task.py

# 不会作为脚本运行，所以可以使用相对导入，此时的scripts包在环境变量中
from .celery import app
# from scripts.celery_task.celery import app

@app.task
def hello_task():
    print('伪代码，随便写的')
    return '没有更新'

User_task.py

from .celery import app
# from scripts.celery_task.celery import app


@app.task
def send_sms(phone):
    print('手机号：%s，发送成功' % phone)
    return True

celery.py

from celery import Celery
import time

# 结果存储
# backend='redis://:密码@127.0.0.1:6379/2' # 有密码的写法
backend = 'redis://127.0.0.1:6379/2'  # 无密码的写法

# 消息中间件
# broker='redis://:123456@127.0.0.1:6379/1' # 有密码写法
broker = 'redis://127.0.0.1:6379/1'  # 无密码写法

app = Celery('test', broker=broker, backend=backend,
             # 以后任务不写在这里了，放到单独的py文件中
             # 写好include，会去相应的py下检索任务，这些任务都被app管理
             include=[
                 'celery_task.home_task',
                 'celery_task.order_task',
                 'celery_task.user_task'
             ])

add_task.py

from celery_task.user_task import send_sms

# 测试路径是否成功：手机号：15058066669，发送成功
# res = send_sms('15058066669')

res = send_sms.apply_async(args=['15058066669'])
print(res)

get_task.py

from celery_task.celery import app  # 借助于app
from celery.result import AsyncResult  # 导入一个类，来查询结果
id = 'f6c16313-8b02-4127-9f15-703e17e78db2'
if __name__ == '__main__':
    res = AsyncResult(id=id, app=app)  # 根据id，去哪个app中找哪个任务，
    if res.successful(): # 执行成功
        result = res.get()
        print('任务执行成功')
        print(result) # 15
    elif res.failed():
        print('任务失败')
    elif res.status == 'PENDING':
        print('任务等待中被执行')
    elif res.status == 'RETRY':
        print('任务异常后正在重试')
    elif res.status == 'STARTED':
        print('任务已经开始被执行')

celery执行任务

celery执行异步任务有两种方式，上面第一种，现在第二种

# 任务名.delay(参数，参数)
# 异步执行

celery执行延迟任务，默认用的是utc时间，东八区

from datetime import datetime, timedelta

eta = datetime.utcnow()  # 现在的时间，不过默认用的是utc时间，
etc = timedelta(seconds=5)  # 5秒后的时间
i = eta + etc  # 现在的时间+10秒
res = send_sms.apply_async(args=['188888'], eta=i) # eta参数表示延迟时间
print(res)

celery执行定时任务

scripts.py/celery_task/celery.py
###### 第一步:在app中写入定时任务
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False
### celery的配置信息---结束###

#### 定时任务
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {
    'send_sms_5': {
        'task': 'celery_task.user_task.send_sms',  # 哪个任务
        'schedule': timedelta(seconds=5),  # 每5s干一次
        # 'schedule': crontab(hour=8, day_of_week=1),  # 每周一早八点
        'args': ('18988377473',),
    },
}


###### 第一步:启动worker
celery -A celery_task worker -l info
####  第三步：启动beat   【【【【注意路径】】】】】
celery -A celery_task beat -l info


### 本质是beat 5s钟提交一次任务，worker执行

django中使用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包必须放置项目的根目录下

举例，异步创建用户：

celery.py

from celery import Celery
import time

"""重点：加载django配置环境,想要使用django的功能必须写这两句话"""
import os
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "luffy_api.setting.dev")

# 结果存储
# backend='redis://:密码@127.0.0.1:6379/2' # 有密码的写法
backend = 'redis://127.0.0.1:6379/2'  # 无密码的写法

# 消息中间件
# broker='redis://:123456@127.0.0.1:6379/1' # 有密码写法
broker = 'redis://127.0.0.1:6379/1'  # 无密码写法

app = Celery('test', broker=broker, backend=backend,
             # 以后任务不写在这里了，放到单独的py文件中
             # 写好include，会去相应的py下检索任务，这些任务都被app管理
             include=[
                 # 'celery_task.home_task'
                 # 'celery_task.order_task'
                 'celery_task.user_task'
             ])

###### 第一步:在app中写入定时任务
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False
### celery的配置信息---结束###

from datetime import timedelta
from celery.schedules import crontab

app.conf.beat_schedule = {
    'send_sms_5': {
        'task': 'celery_task.user_task.send_sms',  # 哪个任务
        'schedule': timedelta(seconds=5),  # 每5s干一次
        # 'schedule': crontab(hour=8, day_of_week=1),  # 每周一早八点
        'args': ('100000000',),
    },
}

user_task.py

# 之所以能上使用Django中的东西因为celery.py导入那两句话
from user.models import User

@app.task
def create_user(mobile,username,password):
    User.objects.create_user(mobile=mobile,username=username,password=password)
    return True

Luffy_api/apps/user/views.py

class TestView(APIView):
    def get(self, request):
        # create_user.delay('13000000','junjienb','junjie123')
        create_user.delay('2131231331','lyfnb','lyf123')
        return Response('用户创建任务已提交')

Luffy_api/apps/user/urls.py

from django.urls import path, include
from rest_framework.routers import SimpleRouter
from .views import UserAPIView, TestView


router = SimpleRouter()
router.register('mobile', UserAPIView, 'mobile')
urlpatterns = [
    path('', include(router.urls)),
    path('', TestView.as_view()),
]

轮播图接口夹缓存

luffy_api/apps/home/views.py

from django.shortcuts import render

# Create your views here.
from .models import Banner
from rest_framework.viewsets import GenericViewSet
from rest_framework.mixins import ListModelMixin
from .serializer import BannerSerializer
from utils.response import APIResponse
from django.conf import settings
from django.core.cache import cache


# 轮播图接口
class BannerView(GenericViewSet, ListModelMixin):
    # 获取所有接口，自动生成路由
    queryset = Banner.objects.filter(is_delete=False, is_show=True).order_by('orders')[:settings.BANNER_COUNT]
    # print(queryset)
    serializer_class = BannerSerializer
    # print(serializer_class)
    """
    希望返回到前端的数据格式
    {code:100,msg:成功，result:[
    {img:地址，link:跳转地址，orders:顺序，title:名字},
    {img:地址，link:跳转地址，orders:顺序，title:名字}]}
    """
    def list(self, request, *args, **kwargs):
        """
        先到redis中查询，如果有，直接返回，没有再执行super()-->到数据库中查询，
        """
        redis_banner_list = cache.get('banner_list')
        if redis_banner_list:
            print('走了缓存')
            # print(redis_banner_list)
            return APIResponse(result=redis_banner_list)
        else:
            print('没走缓存')
            res = super().list(request, *args, **kwargs)
            """
            [OrderedDict([('title', 'banner1'), ('image', 'http://127.0.0.1:8004/media/banner/banner1.png'), ('link', 'http://www.baidu.com'), ('orders', 1)]), 
            OrderedDict([('title', 'banner2'), ('image', 'http://127.0.0.1:8004/media/banner/banner2.png'), ('link', '/home'), ('orders', 2)]), 
            OrderedDict([('title', 'banner3'), ('image', 'http://127.0.0.1:8004/media/banner/banner3.png'), ('link', '/home'), ('orders', 3)])]
            """
            # print(res.data)
            cache.set('banner_list',res.data)
            return APIResponse(result=res.data)

由于Redis中没有缓存，打印没走缓存，并创建了缓存的k:banner_list,v:res.data

此时redis中没有缓存

以后访问轮播图接口就走缓存，不用再前往数据库中查数据，以后如果接口响应慢，第一想法先加缓存：把查出来的数据缓存到redis中，再来请求，先从redis中查，如果没有，再去mysql查，然后在redis缓存一份。

既然使用了缓存就会存在缓存的坑：

假设redis中有一份数据，MySQL中有一份数据。

存在问题：MySQL更新了，Redis没有更新。这里的更新值得是数据更新。

对于这种情况，有个专业名词叫做：双写一致性问题。

解决双写一致性问题，至于使用哪种看使用场景：

• 第一：定时更新，10分钟更新一次缓存。
• 第二：写入MySQL，删除缓存。
• 第三：写入MySQL，更新缓存。

我们写的是定时更新。

celery_task/celery.py

from celery import Celery

# 一、加载django配置环境,想要使用django的功能必须写这两句话
import os

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "luffy_api.setting.dev")

# 结果存储
# backend='redis://:密码@127.0.0.1:6379/2' # 有密码的写法
backend = 'redis://127.0.0.1:6379/2'  # 无密码的写法

# 消息中间件
# broker='redis://:123456@127.0.0.1:6379/1' # 有密码写法
broker = 'redis://127.0.0.1:6379/1'  # 无密码写法

app = Celery('test', broker=broker, backend=backend,
             # 以后任务不写在这里了，放到单独的py文件中
             # 写好include，会去相应的py下检索任务，这些任务都被app管理
             include=[
                 'celery_task.home_task'
             ])

# 第一步:在app中写入定时任务
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False
# celery的配置信息---结束###

from datetime import timedelta

# celery 定时任务的配置
app.conf.beat_schedule = {
    'banner_list_5': {
        'task': 'celery_task.home_task.update_banner_list',  # 哪个任务
        'schedule': timedelta(seconds=5),  # 每5s干一次
        # 'schedule': crontab(hour=8, day_of_week=1),  # 每周一早八点
        'args': (),
    },
}

celery_task/home_task.py

# 不会作为脚本运行，所以可以使用相对导入，此时的scripts包在环境变量中
from .celery import app
from home import models
from home import serializer
from django.conf import settings
from django.core.cache import cache


@app.task
def update_banner_list():
    queryset = models.Banner.objects.filter(is_delete=False, is_show=True).order_by('-orders')[:settings.BANNER_COUNT]
    banner_list = serializer.BannerSerializer(queryset, many=True).data
    # 拿不到request对象，所以头像的连接base_url要自己组装
    for banner in banner_list:
        print(2)
        # 修改为自己的地址
        banner['image'] = 'http://127.0.0.1:8004%s' % banner['image']

    cache.set('banner_list', banner_list, 86400)
    return True

luffy_api/apps/home/views.py

from django.shortcuts import render

# Create your views here.
from .models import Banner
from rest_framework.viewsets import GenericViewSet
from rest_framework.mixins import ListModelMixin
from .serializer import BannerSerializer
from utils.response import APIResponse
from django.conf import settings
from django.core.cache import cache


# 轮播图接口
class BannerView(GenericViewSet, ListModelMixin):
    # 获取所有接口，自动生成路由
    queryset = Banner.objects.filter(is_delete=False, is_show=True).order_by('orders')[:settings.BANNER_COUNT]
    # print(queryset)
    serializer_class = BannerSerializer
    # print(serializer_class)
    """
    希望返回到前端的数据格式
    {code:100,msg:成功，result:[
    {img:地址，link:跳转地址，orders:顺序，title:名字},
    {img:地址，link:跳转地址，orders:顺序，title:名字}]}
    """
    def list(self, request, *args, **kwargs):
        """
        先到redis中查询，如果有，直接返回，没有再执行super()-->到数据库中查询，
        """
        redis_banner_list = cache.get('banner_list')
        if redis_banner_list:
            print('走了缓存')
            # print(redis_banner_list)
            return APIResponse(result=redis_banner_list)
        else:
            print('没走缓存')
            res = super().list(request, *args, **kwargs)
            """
            [OrderedDict([('title', 'banner1'), ('image', 'http://127.0.0.1:8004/media/banner/banner1.png'), ('link', 'http://www.baidu.com'), ('orders', 1)]), 
            OrderedDict([('title', 'banner2'), ('image', 'http://127.0.0.1:8004/media/banner/banner2.png'), ('link', '/home'), ('orders', 2)]), 
            OrderedDict([('title', 'banner3'), ('image', 'http://127.0.0.1:8004/media/banner/banner3.png'), ('link', '/home'), ('orders', 3)])]
            """
            # print(res.data)
            cache.set('banner_list',res.data)
            return APIResponse(result=res.data)