python中celery的基本使用详情

目录

• 1.基本介绍
• 2.使用场景
• 3.工作流程和组成部分
• 4.Celery执行异步任务
• 4.1 基础使用

1.基本介绍

Celery 是由Python 编写的简单，灵活，可靠的用来处理大量信息的分布式系统,它同时提供操作和维护分布式系统所需的工具。Celery 专注于实时任务处理，支持任务调度。

简单的说，它就是一个分布式队列的管理工具，用celery提供的接口快速实现并管理一个分布式的任务队列。

有一点我们需要搞清楚，Celery 本身并不是任务队列，它是一个分布式队列的管理工具，Celery封装好了操作常见任务队列的各种操作，比如说从监听某个任务队列并从该队列中拿到数据进行消费。

2.使用场景

它可以让任务的执行完全脱离主程序，甚至可以被分配到其他主机上运行。我们通常使用它来实现异步任务(async task)和定时任务(crontab)。

• 异步任务： 将耗时操作任务提交给Celery去异步执行，比如发送短信/邮件、消息推送、音视频处理等等
• 定时任务： 定时执行某件事情，比如每天数据统计

3.工作流程和组成部分

这里用一张图片说明下：

Celery的架构由三部分组成，消息中间件（message broker），任务执行单元（worker）和任务执行结果存储（task result store）组成。

消息中间件：

Celery本身不提供消息服务，但是可以方便的和第三方提供的消息中间件集成。包括RabbitMQ, Redis等等，官方推荐用rabbitMQ，因为它持久稳定。

任务执行单元：

Worker是Celery提供的任务执行的单元，worker并发的运行在分布式的系统节点中。

任务结果存储：

Task result store用来存储Worker执行的任务的结果，Celery支持以不同方式存储任务的结果，包括AMQP, redis等

另外， Celery还支持不同的并发和序列化的手段。

并发：Prefork, Eventlet, gevent, threads/single threaded

序列化：pickle, json, yaml, msgpack. zlib, bzip2 compression， Cryptographic message signing 等等 先安装模块

pip install celery
pip install redis

4.Celery执行异步任务

4.1 基础使用

这里项目结构如下：

第一步：先创建celery相关配置配置celery_object.py

import celery

# 执行如下命令: celery -A celery_object worker -l info

backend = "redis://127.0.0.1:6379/4"  # 设置redis的4号数据库来存放结果
broker = "redis://127.0.0.1:6379/5"  # 设置redis的5号数据库存放消息中间件
celery_app = celery.Celery(
    "celery_demo",
    backend=backend,
    broker=broker,
    include=[
        "celery_task",
    ],
)

celery_app.conf.task_serializer = "json"
celery_app.conf.result_serializer = "json"
celery_app.conf.accept_content = ["json"]

celery_app.conf.timezone = "Asia/Shanghai"  # 时区
celery_app.conf.enable_utc = False  # 是否使用UTC

参数说明：

• backend 就是异步任务执行完成以后，结果存放的地方。
• broker 就是具体执行任务的工作节点。
• celery.Celery()方法是实例化一个celery对象。

第二步：创建任务相关的文件celery_task.py

import time

from celery_object import celery_app

@celery_app.task
def send_email(name):
    print("向%s发送邮件..." % name)
    time.sleep(5)
    print("向%s发送邮件完成" % name)
    return f"成功拿到{name}发送的邮件!"

@celery_app.task
def send_msg(name):
    print("向%s发送短信..." % name)
    time.sleep(5)
    print("向%s发送短信完成" % name)
    return f"成功拿到{name}发送的短信!"

通过@celery_app.task这样的装饰器，成功的把对应的函数变成对应celery的异步worker函数。

紧接着我们在项目当前所在的目录执行命令：

celery -A celery_object worker -l info

• -A 指的是application应用对象
• worker 就是工作人（固定写法）
• -l 指的是日志级别，这里是打印info级别的日志

之后就可以有下面的输出显示就代表celery启动成功：

之后我们就可以向celery生产任务了，创建produce_result.py文件。

from celery_task import send_email, send_msg

if __name__ == "__main__":
    for i in range(10):
        result = send_email.delay(f"张三{i}")
        print(result.id)
        result2 = send_msg.delay(f"李四{i}")
        print(result2.id)

运行生产任务的程序，会看到如下的数据，这里打印的就是任务ID。

然后在终端可以看到下面的东西，就代表celery成功的拿到队列中任务 并进行消费了。

然后打开我们的redis可以看到有对应的数据记录。

与此同时 我们还可以查看celery任务ID的状态，check_result.py写入如下：

from celery.result import AsyncResult
from celery_object import celery_app

async_result = AsyncResult(id="d1c722fa-4ebf-432e-967e-a462bdefeac4", app=celery_app)
print("任务状态:", async_result.status)
if async_result.successful():
    result = async_result.get()
    print(result)
    # result.forget() # 将结果删除
elif async_result.failed():
    print("执行失败")
elif async_result.status == "PENDING":
    print("任务等待中被执行")
elif async_result.status == "RETRY":
    print("任务异常后正在重试")
elif async_result.status == "STARTED":
    print("任务已经开始被执行")

运行结果：

任务状态: SUCCESS
成功拿到李四0发送的短信!

到此这篇关于python中celery的基本使用详情的文章就介绍到这了,更多相关python celery内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们！