首页 > 其他分享 >Celery框架的基本使用与介绍

Celery框架的基本使用与介绍

时间:2023-03-08 21:46:31浏览次数:38  
标签:celery task 框架 py 介绍 Celery 任务 print import

目录

Celery介绍、安装、基本使用

一、Celery服务

什么是Celery:

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

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

Celery的运行原理:

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

'''

1、celery架构

  • 消息中间件:broker

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

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

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

img

使用场景:

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

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('任务已经开始被执行')

3、celer包结构【使用包写一个小游戏】

什么是包结构:通过将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命令
	

- 5、第五步:提桥任务:
    # 导入任务
    from celery_bag.celery_task.task import send_sms, add

    # 提交任务
    def add_func(a, b):
        return add.delay(a, b)


    def send_func(mobile, code):
        return send_sms.delay(mobile, code)
    
 - 6、触发任务提交:
    import os
    import redis

    from add_task import add_func, send_func
    from get_result import res_func

    POOL = redis.ConnectionPool(max_connections=100)
    conn = redis.Redis(connection_pool=POOL)

    func_list = {
        '1': add_func,
        '2': send_func
    }

    if __name__ == '__main__':
        while True:
            print('''
            1、异步计算器
            2、模拟发送短信
            3、查看任务执行状态
            4、开启worker,并查看任务执行结果(需要重启系统)
            ''')
            user_choice = input('欢迎来到celery测试系统,请输入您需要执行的功能编号>>>:').strip()
            while user_choice in func_list.keys():
                if user_choice == '1':
                    print('已进入异步计算器功能!')
                    a = '请输入数字 1 '
                    b = '请输入数字 2 '
                else:
                    print('您已进入模拟发送短信功能')
                    a = '手机号'
                    b = '短信内容 '
                    args_1 = input('请输入%s>>>:' % a).strip()
                    args_2 = input('请输入%s>>>:' % b).strip()
                    args_1 = int(args_1)
                    args_2 = int(args_2)
                    res = func_list.get(user_choice)(args_1, args_2)
                    input('''
                        任务提交完成!
                        请记录本次任务ID:%s
                        任意键返回上一层
                               ''' % res)
                    break
                    if user_choice == '3':
                        id = input('请输入任务ID>>>:')
                        res = res_func(id)
                        print(res)
                        continue
                        while user_choice == '4':
                            try:
                                os.system('CD D:\djangoProject\luffy_api\celery_bag\celery_task')
                                os.system('celery  -A  celery_task worker -l info -P eventlet')
                            except Exception as e:
                                print('开始失败,出现错误,请重启系统')
                                print(str(e))
                                break
                                print('自动返回上一层')
                                break
                            else:
                                print('编号输入有误,请重写输入')
                                continue

                                
- 7、第七步:查看任务执行结果:
    # 导入celery实例
    from celery_task.celery import app
    from celery.result import AsyncResult


    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 '任务已经开始被执行,请稍后查询'

标签:celery,task,框架,py,介绍,Celery,任务,print,import
From: https://www.cnblogs.com/kangssssh/p/17196354.html

相关文章

  • celery
    celery介绍是一个python的框架。主要用来解决定时任务异步任务延迟任务的框架官方文档https://docs.celeryq.dev/en/stable/运行原理1可以不依赖任何服务器,通......
  • 路飞-day9——redis之列表类型、redis之hash类型、redis其他方法、redis管道、django
    目录一、redis之列表(List)类型二、redis之hash(字典)类型三、redis其他方法(所有类型通用的方法)四、redis管道五、django中使用redis方式一方式二方案一方案二六、celery介绍......
  • (P01)C++介绍
    文章目录​​1.需要掌握的重要练习​​​​2.为什么要学习C++​​​​3.C++为什么难学​​​​4.C++11值得学习的新特性​​​​5.几本推荐学习C++的书​​​​6.开发工具......
  • ServletConfig接口介绍
    前言:Servlet容器初始化Servlet时,会为这个Servlet创建一个ServletConfig对象,并将ServletConfig对象作为参数传递给Servlet。通过ServletConfig对象即......
  • celery
    celery介绍和安装#Celery是什么-翻译过来是芹菜的意思,跟芹菜没有关系-框架:服务,python的框架,跟django无关-能用来做什么-1异步任务......
  • ES6-ES11 Map的介绍与API
    <!DOCTYPEhtml><htmllang="en"><head><metacharset="UTF-8"><metaname="viewport"content="width=device-width,initial-scale=1.0"><title>Map</t......
  • Python实战项目-9 Redis/celery-基础使用
    Redis介绍与安装Redis->缓存数据库【大部分时间用来做缓存,不仅仅可以做缓存】也是称为非关系型数据库,区别与Mysql关系型数据库-noSql:泛指非关系型数据库,notonlySql......
  • 虚拟化系列-VMware篇-Horizon介绍
    虚拟化系列-VMware篇-Horizon介绍徐建强发布时间:   2021-04-0814:49:23浏览量:   1074评论:   11点赞:   11云桌面网络应用服务存储计算......
  • 亚马逊RedShift调研-浅谈技术框架、优缺点及网上风评
    RedShift是亚马逊开发的一个站式云上数据仓库服务。网上案例与风评较少,做的一些个人总结。1.RedShift优势及特色1.1官方介绍①借助AmazonRedshift的数据仓库......
  • delphi 报表控件介绍
    报表控件介绍FastReportVCL概述用于生成报告和文档的VCL组件集。它提供可视化模板设计器、对最流行数据源的访问、报告引擎、预览、导出过滤器到30多种格式,以及部署......