VB.net 2010 视频教程 VB.net 2010 视频教程 python基础视频教程
SQL Server 2008 视频教程 c#入门经典教程 Visual Basic从门到精通视频教程
当前位置:
首页 > temp > python入门教程 >
  • Celery框架从入门到精通

Celery介绍、安装、基本使用

一、Celery服务

什么是Celery:

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

  • Celery可以用来做什么:
    • 异步任务
    • 定时任务
    • 延迟任务

Celery的运行原理:

  • 可以不依赖任何服务,通过自身命令,启动服务
  • celery服务为其他项目服务提供异步解决任务需求
# 注:会有两个服务同时运行
    - 项目服务
    - celery服务
    项目服务将需要异步处理的任务交给celery服务,celery就会在需要时异步完成项目的需求
    
    
'''
人是一个独立运行的服务 | 医院也是一个独立运行的服务
	正常情况下,人可以完成所有健康情况的动作,不需要医院的参与;但当人生病时,就会被医院接收,解决人生病问题
	人生病的处理方案交给医院来解决,所有人不生病时,医院独立运行,人生病时,医院就来解决人生病的需求

'''

1、celery架构

  • 消息中间件:broker

    • 提交的任务【函数】都放在这里, celery本身不能提供消息中间件
    • 需要借助于第三方: redis或rabbitmq
  • 任务执行单元:worker

    • 真正执行任务的的地方,一个个进程中执行函数
  • 结果储存:backend

    • 函数return的结果都存储在这里, celery本身不提供结果存储
    • 需要借助于第三方: redis或rabbitmq

使用场景:

  • 异步执行:解决耗时任务
  • 延迟执行:解决延迟任务
  • 定时执行:解决周期任务

2、celery快速使用

Celery不支持在windows上直接运行,通过eventlet支持在win上运行

安装:

pip install celery
pip install eventlet  # windows需要安装 

快速使用:

- 1、第一步:创建一个py文件(main.py),用于实例化celery对象,编写需要执行的函数
    # 1、导入模块
    from celery import Celery

    # 2、指定briker,用于存放提交的异步任务
    broker = 'redis://127.0.0.1:6379/1'
    # 3、指定backend,用于存放函数执行结束的结果
    backend = 'redis://127.0.0.1:6379/2'
    # 实例化celery对象
    app = Celery('test', broker=broker, backend=backend)


    # 编写一个函数,装饰上celery对象
    @app.task
    def add(a, b):
        import time
        time.sleep(3)
        print('add函数执行完成')
        return a + b
       
- 2、第二步:再次创建一个py文件(run.py),用于将函数提交给celery
    # 1、导入刚才编写的函数
    from main import add

    # 2、将任务提交给broker,函数需要的参数需要传入
    res = add.delay(1, 2)
    # 3、提交后可以获得该任务的ID,可通过ID可以查询任务执行结果
    print(res)  # 0213d2c2-453e-41a8-a171-e31f1f2f4883
    
  
- 3、第三步:使用命令开启worker (也可以提前开启,任务提交后就会直接执行)
	# 启动worker命令,win需要安装eventlet
	# 启动需要进入main.py文件的目录下
	win:
		-4.x之前版本
			celery worker -A main -l info -P eventlet
		-4.x之后
			celery  -A main  worker -l info -P eventlet
 	mac:
		celery  -A main  worker -l info

- 4、第四步:worker会将执行的结果存在之前指定的broker目录下(指定的redis数据库)
	
- 5、第五步:通过代码查看执行结果(创建新的py文件,专门用于查看执行结果)
    # 1、导入celery实例的对象
    from main import app
    # 2、导入该模块用于查看结果
    from celery.result import AsyncResult

    # 3、将提交的任务编号拿过来,用于查询结果
    id = '0213d2c2-453e-41a8-a171-e31f1f2f4883'

    # 4、指定该文件为启动文件
    if __name__ == '__main__':
        # 实例化对象,将任务的ID和celery实例化对象当作参数传入
        a = AsyncResult(id=id, app=app)
        # 判断执行结果
        if a.successful():  # 执行完了
            result = a.get()
            print(result)
        elif a.failed():
            print('任务失败')
        elif a.status == 'PENDING':
            print('任务等待中被执行')
        elif a.status == 'RETRY':
            print('任务异常后正在重试')
        elif a.status == 'STARTED':
            print('任务已经开始被执行')

二、Celer包结构

1、创建clery包结构

什么是包结构:通过将celery服务封装成包的形式,放在项目需要使用的时候导入即可

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

创建包:

创建一个包,名为:celery_task

- 1、第一步:在包下创建py文件(名字必须为celery.py)
    # 导入celery模块
    from celery import Celery
    # 导入配置broker和backend
    from .settings import BACKEND, BROKER

    # 实例化celery对象
    app = Celery('test',
                 broker=BROKER, 
                 backend=BACKEND,
                 include=['celery_task.order_task', 
                          'celery_task.user_task'])

- 2、第二步:创建settings.py,用于存放配置
    BROKER = 'redis://127.0.0.1:6379/1'
    BACKEND = 'redis://127.0.0.1:6379/2'
    
- 3、第三步,创建py文件(task.py),用于存放需要执行的异步任务
    # 导入celery实例对象
    from .celery import app

    # 计算函数
    @app.task()
    def add(a, b):
        print('计算结果为:', a + b)
        return True

    # 模拟发送短信
    @app.task()
    def send_sms(mobile, code):
        print('已向手机号:%s 发送短信,验证码为:%s' % (mobile, code))
        return True
    
- 4、第四步:开启worker
	切换到celery所在的目录下,开启worker命令
	celery -A celery_task worker -l info -P eventlet
	

- 5、第五步:提桥任务: # add_task.py 文件下
    # 提交任务,这里模拟的是异步任务的提交
    res = add.delay(a, b)  # 提交后可以接收任务的ID
    res1 = send_sms.delay(mobile, code)
    
                                
- 6、第六步:查看任务执行结果: # get_result.py 文件下
    # 导入celery实例
    from celery_task.celery import app
    from celery.result import AsyncResult
    
	 id = res
    id1 = res1
		
    # 通过传入任务的ID就可以查询到任务的执行结果
    def res_func(id):
        id = id
        a = AsyncResult(id=id, app=app)
        if a.successful():  # 执行完了
            result = a.get()
            if result: return '执行完成'
        elif a.failed():
            return '任务失败,失败的原因可能是未开启worker'
        elif a.status == 'PENDING':
            return '任务等待中被执行,当前任务较多或未开启worker'
        elif a.status == 'RETRY':
            return '任务异常后正在重试'
        elif a.status == 'STARTED':
            return '任务已经开始被执行,请稍后查询'

2、Celery执行异步任务、延迟任务、定时任务

执行异步任务:

# 代码用法:
	函数名.delay('函数执行需要的参数')
	res = func.delay(*args,**kwargs)   # res 用于接收提交任务的ID

执行延迟任务:

# 代码用法:
    # 1、执行延迟任务
    from datetime import datetime, timedelta

    # 设置延迟后的时间,一分钟后执行
    eat = datetime.utcnow() + timedelta(minutes=1)

    # 提交任务
    res = send_sms.apply_async(args=['13855411111', '123'], eta=eta)

执行定时任务:

执行定时任务需要启动beat和worker

  • beat:定时提交任务的进程---》配置在app.conf.beat_schedule的任务
  • worker:执行任务
- 第一步:在celery的py文件中写入
    # 导入定时需要的模块
    from celery.schedules import crontab
    # 第一步:在celery的py文件中写入
    app.conf.timezone = 'Asia/Shanghai'
    # 是否使用UTC
    app.conf.enable_utc = False
    # celery的配置文件#####
    # 任务的定时配置
    app.conf.beat_schedule = {
        'send_sms': {  # 配置执行函数的名字
            'task': 'celery_task.task.send_sms',  # 导入任务的位置
            # 'schedule': timedelta(seconds=3),  # 时间对象
            # 'schedule': crontab(hour=8, day_of_week=1),  # 每周一早八点
            'schedule': crontab(hour=9, minute=43),  # 每天9点43
            'args': ('18888888', '6666'),  # 配置执行函数需要的参数
        },
    }

    
- 第二步:启动beat  # 启动后配配置的任务会自动提交
	celery -A celery_task beat -l info
    
- 第三步:启动worker  # beat提交的任务被被执行
	celery -A celery_task worker -l info -P eventlet
    

三、Django中使用celery

补充:

如果在公司中,只做定时任务有一个框架更简单一点

  • APSchedule:https://blog.csdn.net/qq_41341757/article/details/118759836

使用步骤:

-1 把咱们写的包,复制到项目目录下
    -luffy_api
        -celery_task #celery的包路径
        -luffy_api  #源代码路径

-2 在使用提交异步任务的位置,导入使用即可
    -视图函数中使用,导入任务
    -任务.delay()  # 提交任务


-3 启动worker,如果有定时任务,启动beat

-4 等待任务被worker执行

-5 在视图函数中,查询任务执行的结果

1、模拟写一个异步秒杀任务

后端

view.py

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


# 秒杀接口
class SeckillView(ViewSet):

    # 开启秒杀
    @action(methods=['GET'], detail=False)
    def seckill(self, request):
        # 获取商品链接
        goods_id = request.query_params.get('goods_id')
        # 将任务提交给worker
        res = sckill_task.delay(goods_id)
        # 将任务的ID反馈给前端
        return APIResponse(task_id=str(res))

    # 查询秒杀结果
    @action(methods=['GET'], detail=False)
    def get_result(self, request):
        # 前端将任务ID产过来,用于接收结果
        task_id = request.query_params.get('task_id')
        # 调用接口,查询结果
        a = AsyncResult(id=task_id, app=app)
        if a.successful():
            result = a.get()
            if result:
                return APIResponse(msg='秒杀成功')
            else:
                return APIResponse(code=101, msg='手速满了,秒杀失败')
        elif a.status == 'PENDING':
            return APIResponse(code=666, msg='加速秒杀中')
        return APIResponse(msg='错误')

celery.py ---->秒杀任务

import random


# 秒杀函数
@app.task()
def sckill_task(goods_id):
    print('商品正在秒杀中')
    time.sleep(random.choice([6, 7, 8, 9]))
    print('商品秒杀结束')
    return random.choice([True, False])

前端:

<template>
  <div>
    <button @click="clickHandle">点击秒杀</button>
  </div>
</template>

<script>
export default {
  name: "Template",
  data() {
    return {
      // 用于接收任务ID
      task_id: '',
      // 用户存放定时任务
      t: ''
    }
  },
  methods: {
    // 用户点击秒杀后发送请求
    clickHandle() {
      // 向厚点提交秒杀任务
      this.$axios.get(this.$settings.BASE_URL + '/user/seckill/seckill/?goods_id=1').then(res => {
        // 判断任务是否提交成功
        if (res.data.code == 100) {
          // 提交成功会获取到任务ID
          this.task_id = res.data.task_id
          // 告知用户商品正在秒杀中
          this.$message('正在秒杀中')
          // 启动一个定时任务,每隔3秒向后端发送请求,获取任务是否提交成功
          this.t = setInterval(res => {
            // 定时向后端发送请求,判断秒杀结果
            this.$axios.get(this.$settings.BASE_URL + '/user/seckill/get_result/?task_id=' + this.task_id).then(res => {
              // 判断任务是否结束
              if (res.data.code == 666) {
                this.$message(res.data.msg)
                // 任务结束反馈结果,关闭定时器
              } else {
                this.$message(res.data.msg)
                // 关闭定时器
                clearInterval(this.t)
                this.t = ''
              }
            })
          }, 3000)
        }
      })
    }
  }
}
</script>

2、总结

  • 第一步:将celery包复制到项目路径下
-luffy_api
    -celery_task #celery的包路径
    celery.py  # 一定不要忘了一句话
        import os
         # 重点:celery中使用djagno,任务中可能会使用django的orm,缓存,表模型。。。。一定要加
        os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'luffy_api.settings.dev')
        -luffy_api  #源代码路径
  • 第二步:在需要使用异步的地方导入celery实例即可使用
-视图函数中使用,导入任务
-任务.delay()  # 提交任务
  • 第三步:启动worker,如果有定时任务,启动beat

  • 第四步: 等待任务被worker执行

  • 第五步:在视图函数中,查询任务结果

    出处:https://www.cnblogs.com/kangssssh/p/17201349.html
     


相关教程