Python 强大的任务调度框架 Celery！

# 文件名：app.pyimport time# 这个 Celery 就类似于 flask.Flask# 然后实例化得到一个appfrom celery import Celery# 指定一个 name、以及 broker 的地址、backend 的地址app = Celery( "satori", # 这里使用我服务器上的 Redis # broker 用 1 号库, backend 用 2 号库 broker="redis://:maverick@82.157.146.194:6379/1", backend="redis://:maverick@82.157.146.194:6379/2")# 这里通过 @app.task 对函数进行装饰# 那么之后我们便可调用 task.delay 创建一个任务@app.taskdef task(name, age): print("准备执行任务啦") time.sleep(3) return f"name is {name}, age is {age}"

import timefrom app import task# 从 app 导入 task, 创建任务, 但是注意: 不要直接调用 task# 因为那样的话就在本地执行了, 我们的目的是将任务发送到队列里面去# 然后让监听队列的 worker 从队列里面取任务并执行# 而 task 被 @app.task 装饰, 所以它不再是原来的 task 了# 我们需要调用它的 delay 方法# 调用 delay 之后, 就会创建一个任务# 然后发送到队列里面去, 也就是我们这里的 Redis# 至于参数, 普通调用的时候怎么传, 在 delay 里面依旧怎么传start = time.perf_counter()task.delay("古明地觉", 17)print( time.perf_counter() - start)# 0.11716766700000003

from app import taskres = task.delay("古明地觉", 17)print(type(res))"""<class>"""# 直接打印，显示任务的 idprint(res)"""4bd48a6d-1f0e-45d6-a225-6884067253c3"""# 获取状态, 显然此刻没有执行完# 因此结果是PENDING, 表示等待状态print(res.status)"""PENDING"""# 获取 id，两种方式均可print(res.task_id)print(res.id)"""4bd48a6d-1f0e-45d6-a225-6884067253c34bd48a6d-1f0e-45d6-a225-6884067253c3"""# 获取任务执行结束时的时间# 任务还没有结束, 所以返回Noneprint(res.date_done)"""None"""# 获取任务的返回值, 可以通过 result 或者 get()# 注意: 如果是 result, 那么任务还没有执行完的话会直接返回 None# 如果是 get(), 那么会阻塞直到任务完成print(res.result)print(res.get())"""Nonename is 古明地觉, age is 17"""# 再次查看状态和执行结束时的时间# 发现 status 变成SUCCESS# date_done 变成了执行结束时的时间print(res.status)# 但显示的是 UTC 时间print(res.date_done)"""SUCCESS2022-09-08 06:40:34.525492"""</class>

import time# 我们不光要导入 task, 还要导入里面的 appfrom app import app, task# 导入 AsyncResult 这个类from celery.result import AsyncResult# 发送任务到队列当中res = task.delay("古明地觉", 17)# 传入任务的 id 和 app, 创建 AsyncResult 对象async_result = AsyncResult(res.id, app=app)# 此时的这个 res 和 async_result 之间是等价的# 两者都是 AsyncResult 对象, 它们所拥有的方法也是一样的# 下面用谁都可以while True: # 等价于async_result.state == "SUCCESS" if async_result.successful(): print(async_result.get()) break # 等价于async_result.state == "FAILURE" elif async_result.failed(): print("任务执行失败") elif async_result.status == "PENDING": print("任务正在被执行") elif async_result.status == "RETRY": print("任务执行异常正在重试") elif async_result.status == "REJECTED": print("任务被拒绝接收") elif async_result.status == "REVOKED": print("任务被取消") else: print("其它的一些状态") time.sleep(0.8)"""任务正在被执行任务正在被执行任务正在被执行任务正在被执行name is 古明地觉, age is 17"""