01. celery

celery

一、celery简介

1 celery:芹菜（跟芹菜没有任何关系）
2 python中的一个分布式异步任务框架
	-执行异步任务---(对立：同步任务):解决耗时任务,将耗时操作任务提交给Celery去异步执行，比如发送短信/邮件、消息推送、音视频处理等等
    -执行延时任务（5分钟后干一件事）:解决延迟任务
    -执行定时任务：每天，隔几分钟，干什么事:解决周期(周期)任务,比如每天数据统计
3 解释
Celery is a project with minimal funding, so we don’t support Microsoft Windows. Please don’t open any issues related to that platform.

4 celery特点（了解）
    1）可以不依赖任何服务器，通过自身命令，启动服务(内部支持socket)
    2）celery服务为为其他项目服务提供异步解决任务需求的

5 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.7
(5.x版本后 有的命令不能用)

三、celery基本使用

1、写一个py文件

  • t_celery.py
import celery
# 消息中间件（redis）
broker = 'redis://127.0.0.1:6379/1'  # 1 表示使用redis的db1
# 结果存储（redis）
backend = 'redis://127.0.0.1:6379/2'  # 2表示使用redis的db2
# 实例化得到对象，指定中间件和结果存储
app = celery.Celery('test', broker=broker, backend=backend)

@app.task
def add(a,b):
    return a+b

@app.task
def mul(a,b):
    return a*b

2、提交任务

  • s1.py
from t_celery import add, mul
res = add.delay(4,5)
print(res)  # id号

3、启动worker

# 非windows平台： celery worker -A t_celery -l info
# windows装eventlet：（pip install eventlet） 
Terminal中执行：
celery worker -A t_celery -l info -P eventlet

4、查看执行结果

  • get_res.py
from t_celery import app

from celery.result import AsyncResult

id = 'd6a5a7b6-0166-4267-a4ca-82fb7d8d9a5c'
if __name__ == '__main__':
    res = AsyncResult(id=id, app=app)
    if res.successful():
        result = res.get()
        print(result)
    elif res.failed():
        print('任务执行失败')
    elif res.status == 'PENDING':
        print('任务等待中')
    elif res.status == 'RETRY':
        print('任务异常后正在重试')
    elif res.status == 'STARTED':
        print('任务已经开始内执行')

四、celery多任务结构

package_celery:     # 项目名
    celery_task     # celery包名
        __init__.py 
        celery.py   # celery 的app，必须叫celery
        order_task.py # 任务
        user_task.py  # 任务
    result.py         # 结果查询
    submit_tast.py    # 提交任务

# 启动worker（在package_celery目录下执行）
	celery worker -A celery_task -l info -P eventlet
  
# 提交任务
	from celery_task import order_task, user_task
    # 提交一个给用户发短信的任务
	res = user_task.send_sms.delay('')

1、手动启动文件执行

  • celery.py
import celery

broker = 'redis://127.0.0.1:6379/1'
backend = 'redis://127.0.0.1:6379/2'
app = celery.Celery('test', broker=broker, backend=backend,
                    include=['celery_task.order_task', 'celery_task.user_task'])
  • order_task.py
from .celery import app


@app.task
def process_order(a, b):
    print(a)
    print(b)
    return '订单处理好了'


@app.task
def cancel_order():
    import random
    res = random.choice([1, 0])
    if res == 0:
        print('订单状态改了，取消订单了')
        return True
    else:
        print('订单取消失败')
        return False
  • user_task.py
from .celery import app


@app.task
def send_sms(phone):
    import time
    time.sleep(5)
    print('%s短信发送成功' % phone)
    return '%s短信发送成功' % phone
  • submit_task.py
from celery_task import order_task, user_task
res = user_task.send_sms.delay('13788888888')

print(res)

res1 = order_task.cancel_order.delay()
print(res1)
  • result.py
from celery_task.celery import app
from celery.result import AsyncResult

# submit_task.py 执行的结果：
# 0933f36a-8161-4664-b2ba-8249eef31130
# 2f24c146-4017-4920-a001-83d4c86a8369
id = '2f24c146-4017-4920-a001-83d4c86a8369'
if __name__ == '__main__':
    res = AsyncResult(id=id, app=app)
    if res.successful():
        result = res.get()
        print(result)
    elif res.failed():
        print('任务失败')
    elif res.status == 'PENDING':
        print('任务等待被执行')
    elif res.status == 'RETRY':
        print('任务异常后正在重试')
    elif res.status == 'STARTED':
        print('任务已经开始被执行')
  • 真实应用场景
- 秒杀系统
	- 不能秒超，使用锁（mysql悲观锁，乐观锁），redis锁
        -提高并发量---》把同步做成异步---》使用celery
        	-前端点击秒杀按钮，向后端发送秒杀请求---》同步操作
            
            	-同步操作
                    -请求来到后端，判断数量是否够，如果够，要生成订单（mysql），订单状态是待支付状态					
                    -请求返回，告诉前端，秒杀成功
                    
                -异步操作
                	-请求来到后端，提交一个celery任务---》celery任务异步的执行判断数量是否够，如果够，要生成订单（mysql）
                    -秒杀是否成功的结果还没有，直接返回了（返回任务id）
                    -前端启动一个定时任务，每隔5s，向后台发送一个查询请求，查询秒杀任务是否执行完成（带着任务id查）
                    -如果是未执行状态，或者执行中---》返回给前端，前端不处理，定时任务继续执行
                    -又隔了5s，发送查询，查询到秒杀成功的结果，返回给前端，秒杀成功

2、高级使用之延时任务

  • celery.py

方式一：设定未来的时间发送短信

# eta: 延迟多久执行，需要传utc时间对象
from datetime import datetime
t1 = datetime(2021, 2, 14, 0, 0, 0)
# print(t1)
t2 = datetime.utcfromtimestamp(t1.timestamp())
print(t2)
res = user_task.send_sms.apply_async(args=['13788888888'], eta=t2)  # 在utc时间提交任务

方式二：以当前utc时间往后延迟时间执行

from datetime import datetime, timedelta
ctime = datetime.now()
utc_ctime = datetime.utcfromtimestamp(ctime.timestamp())  # 文件运行的当前utc时间
time_delay = timedelta(seconds=10)  # 延迟时间 
task_time = utc_ctime + time_delay
res = user_task.send_sms.apply_async(args=['13788888888'], eta=task_time)  # 在utc时间提交任务

3、高级使用之定时任务

  • celery.py中配置：
# 时区
app.conf.timezone = 'Asia/Shanghai'
# 是否使用utc时间
app.conf.enable_ut = False

# 任务的定时配置
from datetime import timedelta
from celery.schedules import crontab

app.conf.beat_schedule = {
    'send-msg': {
        'task': 'celery_task.user_task.send_sms',
        'schedule': timedelta(hour=24*10), # 每隔10天执行一次
        # 'schedule': crontab(hour=8, day_of_week=1),  # 每周一早八点
        'args': ('13788888888',),
    }
}

# 启动beat,负责定时提交任务
celery beat -A celery_task -l info
# 启动worker
celery worker -A celery_task -l info -P eventlet

五、django中使用celery

1、celery是独立的，跟框架没有关系
2 django-celery第三方模块，兼容性不好，咱们不采用，咱们使用通用方式
3 目录
	celery_task
    	__init__.py
        celery.py
        home_task.py
        order_task.py
        user_task.py
	luffyapi

1、home_task.py

from celery_task.celery import app
from django.conf import settings
from django.core.cache import cache
from home import models
from home import serializers

@app.task
def update_banner():
    banners = models.Banner.objects.fiter(is_delete=False, is_show=True).order_by('-order')[:settings.BANNER_SIZE]
    ser = serializers.BannerSerializer(instance=banners, many=True)
    banner_data = ser.data
    # celery独立于django项目，所以i头像的地址要自己拼
    for banner in banner_data:
        banner['img'] = 'http://127.0.0.1:8000%s' % banner['img']
    cache.set('banner_data', banner_data)
    return True

2、celery.py

import celery
import os

# 执行django配置文件，环境变量加入
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'luffyapi.settings.dev')

broker = 'redis://127.0.0.1:6379/1'
backend = 'redis://127.0.0.1:6379/2'
app = celery.Celery('test', broker=broker, backend=backend,
                    include=['celery_task.order_task', 'celery_task.user_task', 'celery_task.home_task'])
# 时区
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False

# 任务的定时配置
from datetime import datetime, timedelta
from celery.schedules import crontab

app.conf.beat_schedule = {
    'update-banner': {
        'task': 'celery_task.home_task.update_banner',
        'schedule': timedelta(seconds=60),  # 每隔60秒执行一次
        'args': (),
    },
}

3、views.py

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


def test_celery(request):
    res_id = request.GET.get('id')
    if res_id:
        res = AsyncResult(id=res_id, app=app)
        if res.successful():
            result = res.get()
            print(result)
            return HttpResponse('执行完成了，结果是：%s' % str(result))
    res = user_task.send_sms.delay('15722228888')
    return HttpResponse('task_id: %s' % str(res))

六、首页轮播图定时更新

1 把首页轮播图接口改成: 先去缓存中取，缓存中没有，再去数据库查，并将查到的结果放进缓存中去
补充：
	如果你的接口，请求慢，第一反应就是先使用缓存

2 使用celery定时更新缓存
  • home/serializers.py
from utils.response import ApiResponse
from rest_framework.viewsets import ModelViewSet
from rest_framework import mixins
from . import models, serializers
from django.conf import settings
from django.core.cache import cache


class BannerViewSet(ModelViewSet, mixins.ListModelMixin):
    queryset = models.Banner.objects.filter(is_delete=False, is_show=True).order_by('-order')[:settings.BANNER_SIZE]
    serializer_class = serializers.BannerSerializer

    # 改进： 先从缓存中取，有数据就直接返回；没有就去数据库中查，查到了再将数据放进缓存
    def list(self, request, *args, **kwargs):
        banner_data = cache.get('banner_data')
        if banner_data:
            print('查的缓存')
            return ApiResponse(data=banner_data)
        # 如果缓存中没有，再查数据库，查出来，放到缓存中
        res = super().list(request, *args, **kwargs)
        cache.set('banner_data', res.data)
        print('查了数据库')
        return res

七、celery配置文件参数了解

#有些情况可以防止死锁
CELERYD_FORCE_EXECV=True
# 设置并发worker数量
CELERYD_CONCURRENCY=4
#允许重试
CELERY_ACKS_LATE=True
# 每个worker最多执行100个任务，超过了就会被销毁重新创建，可以防止内存泄漏
CELERYD_MAX_TASKS_PER_CHILD=100
# 超时时间
CELERYD_TASK_TIME_LIMIT=12*30

# 补充 Supervisor： 进程管理工具，python写的，在linux上运行，通过配置，保证进程如果崩了，自动重启
版权声明：本文为weixin_49111957原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接和本声明。