环球热门:Celery框架的基本使用与介绍

2023-03-08 22:03:17 来源：博客园

Celery介绍、安装、基本使用
一、Celery服务
- 1、celery架构
- 2、celery快速使用
- 3、celer包结构【使用包写一个小游戏】

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

关键词：