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

运行结果：