关于后端:celery笔记一之celery介绍启动和运行结果跟踪

本文首发于公众号：Hunter 后端
原文链接：celery 笔记一之 celery 介绍、启动和运行后果跟踪

本篇笔记内容如下：

1. celery 介绍
2. celery 筹备
3. celery 启动和异步工作的运行
4. 运行后果跟踪

1、celery 介绍

celery 大抵有两种利用场景，一种是异步工作，一种是定时工作。

比如说在一个接口申请中，某个函数执行所需的工夫过长，而前端页面并不是立即须要在接口中获取处理结果，能够将这个函数作为异步工作，先返回给前端解决中的信息，在后盾独自运行这个函数，这就是异步工作。

另一个比如说某个函数须要每天晚上运行一遍，不可能人天天守着后盾手动执行一遍这个函数，那么就能够用 celery 来实现这个定时的周期工作。

接下来介绍一下 celery 的组成：

task

这个工作就是咱们后面举的例子的异步工作或者是定时工作，即为 task，咱们能够定义这些工作，而后发送到 broker

broker

broker 能够了解成消息中间件，用于获取异步或者定时工作，造成一个或多个音讯队列，而后发送给 worker 解决这些音讯

broker 的模式能够是 Redis，RabbitMQ 或者其余，这里咱们应用 Redis 作为消息中间件

worker

worker 是解决音讯的程序，获取 broker 中的音讯，而后在 worker 中执行，而后依据配置决定将处理结果发送到 backend

result_backend

在 worker 解决完音讯之后会有 return 或者没有返回后果，都会依据配置将后果发送进去，能够配置成发送到 redis 中，也能够将之存储到 database 中

beat

次要用于调用定时工作，依据设定好的定时工作，比方每天晚上十点执行某个函数，beat 则会在相应的工夫将这个 task 发送给 broker，而后 worker 获取工作进行解决

定时工作除了说的每天晚上十点这种周期工作，也能够是距离工作，比如说每隔多少秒，多少分钟执行一次

留神：异步工作的发送是不通过 beat 解决，间接发送给 broker 的

在下面的构造中，broker 须要将相应的服务比方 redis 运行起来，而 worker 和 beat 须要在手动用程序运行，而且每次更改了定时策略之后须要重新启动 beat 和 worker 能力失效。

2、celery 筹备

接下来咱们实现一个最简略的异步工作，在执行异步工作前，咱们做如下的筹备工作

1. 装置依赖

咱们须要装置一下 celery 和 redis 的依赖：

pip3 install celery==5.1.2 -i https://mirrors.aliyun.com/pypi/simple/
pip3 install redis==3.5.3 -i https://mirrors.aliyun.com/pypi/simple/

2. 消息中间件

这里咱们用到的消息中间件是 redis，能够去官网下载一个 redis 启动，也能够应用 docker 来执行装置。

我在之前的 docker 系列笔记中有介绍过如何拉取镜像和运行容器，咱们这里间接应用 docker 来运行：

docker run -itd -p 6379:6379 redis:latest

3. 异步工作筹备

咱们筹备一个最简略的 add() 函数，放在 tasks.py 文件中：

# tasks.py
from celery import Celery

app = Celery('tasks', broker='redis://localhost/0', backend='redis://localhost/1')

@app.task
def add(x, y): 
    return x + y

在这段代码里，咱们引入 Celery 模块，并将其实例化为 app，且配置了 broker 参数，示意音讯队列都会被放在 redis 的第一个数据库下

指定的 backend 参数则示意函数运行的后果被放在 redis 的第二个数据库下

而后用 @app.task 润饰 add 函数，示意它是 app 下的 task 工作

以上，咱们的筹备工作就实现了，接下来尝试运行这个异步工作

3、celery 启动和异步工作的运行

说是 celery 的启动，其实是 worker 的启动，中间件是 redis，曾经在后面的步骤中启动了。

咱们在 tasks.py 所在的文件夹下执行上面的命令：

celery -A tasks worker -l INFO

在这里，tasks 是咱们工作所在的文件名，worker 示意启动的是 worker 程序

-l INFO 则会在控制台打印出 worker 接管到的音讯详情，如果不执行，则信息流不会被打印进去

执行了下面的程序后，能够看到控制台会输入上面这种信息：

 -------------- celery@localhost v5.1.2 (sun-harmonics)
--- ***** ----- 
-- ******* ---- Darwin-21.4.0-x86_64-i386-64bit 2022-07-17 23:56:09
- *** --- * --- 
- ** ---------- [config]
- ** ---------- .> app:         tasks:0x7fc8ddf3df98
- ** ---------- .> transport:   redis://localhost:6379/0
- ** ---------- .> results:     disabled://
- *** --- * --- .> concurrency: 12 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** ----- 
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery
                

[tasks]
  . tasks.add

[2022-07-17 23:56:09,685: INFO/MainProcess] Connected to redis://localhost:6379/0
[2022-07-17 23:56:09,699: INFO/MainProcess] mingle: searching for neighbors
[2022-07-17 23:56:10,737: INFO/MainProcess] mingle: all alone
[2022-07-17 23:56:10,780: INFO/MainProcess] celery@localhost ready.

则示意 worker 启动胜利

执行异步工作

在另一个 shell 窗口，进入 python 的交互界面，输出以下命令：

from tasks import add
res = add.delay(1,2)

add 是咱们须要执行的异步工作的函数名

delay 是异步工作执行的特有办法，这个其实是 apply_async() 函数的简便写法，不带任何参数，apply_async() 除了能够实现异步工作的性能，还能够指定多少工夫后执行，比如说二十秒后执行，这个在前面的笔记中咱们再介绍。

而异步工作的返回咱们这里用 res 来定义，它是一个蕴含了这个工作所有执行信息对象，有工作状态（是否执行胜利），有返回后果（add() 函数的 return），还有这个 task 特有的标识 id 等信息

至此，咱们的一个异步工作的执行就实现了，咱们能够在下一步查看它的运行后果等信息。

4、运行后果跟踪

接下来，咱们在 tasks.py 中建设上面几个函数，来测试咱们对后果的跟踪：

# tasks.py

import time
from celery import Celery

app = Celery('tasks', broker='redis://localhost/0', backend='redis://localhost/1')


@app.task
def add(x, y):
    return x + y


@app.task
def div(x, y):
    return x / y


@app.task
def test_not_finished():
    time.sleep(30)
    return True

而后从新运行 worker：

celery -A tasks worker -l INFO

而后引入和执行函数：

from tasks import add, div, test_not_finished

获取延时工作的后果

res = add.delay(1, 2)
print(res.result)

# 也能够应用 get() 
print(res.get())

get() 函数也能够加个超时的设置：

res.get(timeout=2)

然而这样须要留神，因为如果超时了还未获取到后果，程序就会报错

判断函数运行是否实现

print(res.ready())

打印出的后果为 True 则示意函数运行实现

咱们能够测试函数为实现的状态：

res2 = test_not_finished.delay()

在这个函数里，咱们设置了 30s 的休眠，所以在 30s 內咱们打印后果能够看到 res2.ready() 是为 False 的：

print(res2.ready())

获取 task id

每个被执行的 task 都有各自对应的 id 作为它们的惟一键：

print(res.id)

查看工作执行的状态

# 工作执行是否失败，返回 布尔型数据
is_failed = res.failed()

# 工作执行是否胜利，返回布尔型数据
is_successful = res.successful()

# 执行的工作所处的状态
state = res.state
# state 的值会在 PENDING，STARTED，SUCCESS，RETRY，FAILURE 这几种状态中，别离是 待处理中，工作曾经开始，胜利，重试中，失败

报错解决

如果执行的延时工作在程序中报错，比方咱们定义的 div() 函数，咱们传入的除数为 0 的话，在程序中是会报错的，咱们应用 get() 来获取后果的话程序是会报错的：

res3 = div.delay(3, 0)
res3.get()

# 返回会报错

然而咱们能够应用 propagate=False 参数来疏忽程序的报错:

res3.get(propagate=False)

这样咱们获取的就不是程序报错，而是程序报错的信息作为后果返回

应用 res3.state 发现返回的后果是 FAILURE

当延时工作在程序中报错，它的返回值就不会是正确的，咱们能够通过 res3.traceback 是否有值来判断函数运行过程中是有报错：

if res3.traceback:
    print("延时工作报错")
else:
    print("程序失常运行，能够获取返回值")

result 资源开释

因为 backend 会应用资源来保留和传输后果，为了确保资源被开释，所以在执行完异步工作后，你必须对每一个后果调用 get() 或者 forget() 函数

result.get() 函数获取后果

result.forget() 在 backend 删掉该数据

在官网文档上，意思是 get() 和 forget() 办法都能够开释资源，然而通过我测试，貌似只有 forget() 函数会开释资源

查看是否资源被开释也很简略，登录到对应的 backend，我这里是 redis，应用 redis-cli 或者通过 docker 进入 redis：

select 1

keys*

查看相应的 task id 是否还在列表就能够晓得该资源是否被开释

如果不想手动开释资源，能够在配置里设置一个过期工夫，那么后果就会在指定时间段后被开释：

app.conf.update(result_expires=60)

这个咱们能够在前面的配置里再具体介绍。

以上就是本篇笔记全部内容，下一篇笔记咱们将介绍如何建设一个 celery 我的项目、配置的几种办法及一些根本的配置。

如果想获取更多后端相干文章，可扫码关注浏览：