Celery框架从入门到精通

爱若成殇

目录

• Celery介绍、安装、基本使用
  
• 一、Celery服务
  
  
  
  • 1、celery架构
    
  • 2、celery快速使用
    
  
  
• 二、Celer包结构
  
  
  
  • 1、创建clery包结构
    
  • 2、Celery执行异步任务、延迟任务、定时任务
    
  
  
• 三、Django中使用celery
  
  
  
  • 1、模拟写一个异步秒杀任务
    
  • 2、总结
    
  
  

Celery介绍、安装、基本使用

一、Celery服务

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

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

Celery的运行原理:

• 可以不依赖任何服务，通过自身命令，启动服务
  
• celery服务为其他项目服务提供异步解决任务需求
  

1. # 注：会有两个服务同时运行    - 项目服务    - celery服务    项目服务将需要异步处理的任务交给celery服务，celery就会在需要时异步完成项目的需求        '''人是一个独立运行的服务 | 医院也是一个独立运行的服务        正常情况下，人可以完成所有健康情况的动作，不需要医院的参与；但当人生病时，就会被医院接收，解决人生病问题        人生病的处理方案交给医院来解决，所有人不生病时，医院独立运行，人生病时，医院就来解决人生病的需求'''

复制代码

1、celery架构

• 消息中间件：broker
  
  
  
  • 提交的任务【函数】都放在这里， celery本身不能提供消息中间件
    
  • 需要借助于第三方： redis或rabbitmq
    
  
  
• 任务执行单元：worker
  
  
  
  • 真正执行任务的的地方，一个个进程中执行函数
    
  
  
• 结果储存：backend
  
  
  
  • 函数return的结果都存储在这里， celery本身不提供结果存储
    
  • 需要借助于第三方: redis或rabbitmq
    
  
  

使用场景：

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

2、celery快速使用

Celery不支持在windows上直接运行，通过eventlet支持在win上运行
安装：

1. pip install celerypip install eventlet  # windows需要安装

复制代码

快速使用：

1. - 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服务封装成包的形式，放在项目需要使用的时候导入即可

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

复制代码

创建包：
创建一个包，名为：celery_task

1. - 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执行异步任务、延迟任务、定时任务

执行异步任务：

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

复制代码

执行延迟任务：

1. # 代码用法：    # 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：执行任务
  

1. - 第一步：在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. -1 把咱们写的包，复制到项目目录下    -luffy_api        -celery_task #celery的包路径        -luffy_api  #源代码路径-2 在使用提交异步任务的位置，导入使用即可    -视图函数中使用，导入任务    -任务.delay()  # 提交任务-3 启动worker，如果有定时任务，启动beat-4 等待任务被worker执行-5 在视图函数中，查询任务执行的结果

复制代码

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

后端
view.py

1. from celery.result import AsyncResultfrom celery_task.celery import appfrom 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 ---->秒杀任务

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

复制代码

前端：

1.       点击秒杀  

复制代码

2、总结

• 第一步：将celery包复制到项目路径下
  

1. -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实例即可使用
  

1. -视图函数中使用，导入任务-任务.delay()  # 提交任务

复制代码

• 第三步：启动worker，如果有定时任务，启动beat
  
• 第四步： 等待任务被worker执行
  
• 第五步：在视图函数中，查询任务结果
  

来源:https://www.cnblogs.com/kangssssh/p/17201349.html
免责声明：由于采集信息均来自互联网，如果侵犯了您的权益，请联系我们【E-Mail:cb@itdo.tech】 我们会及时删除侵权内容，谢谢合作！