Celery适用于处理耗时任务，如发送邮件、处理视频等，通过消息队列实现异步执行和负载均衡；使用Flower可监控任务状态，支持重试、错误处理和死信队列应对任务失败。
Celery是一个强大的分布式任务队列，简单来说，它让你能够把一些耗时的操作（比如发送邮件、处理上传的视频）放到后台去执行，而不用阻塞你的Web应用。这样，用户就能更快地得到响应，体验也就更好了。
Celery通过消息队列（通常是rabbitmq或redis）来传递任务，让你的任务可以在不同的服务器上运行，从而实现负载均衡。
解决方案：
首先，你需要安装Celery及其依赖：
这里我选择Redis作为消息代理，你也可以选择RabbitMQ。
然后，创建一个Celery实例，通常在一个单独的
celery.py
文件中：
# celery.py from celery import Celery app = Celery('my_project', broker='redis://localhost:6379/0', # Redis作为消息代理 backend='redis://localhost:6379/0', # Redis作为结果存储 include=['my_project.tasks']) # 包含任务模块 # 可选配置 app.conf.update( result_expires=3600, # 任务结果过期时间 ) if __name__ == '__main__': app.start()
接下来，定义你的异步任务，例如在一个
my_project/tasks.py
文件中：
# my_project/tasks.py from celery import shared_task import time @shared_task def add(x, y): # 模拟耗时操作 time.sleep(5) return x + y
要运行Celery worker，你需要打开一个终端，进入你的项目目录，然后执行：
celery -A my_project worker -l info
-A my_project
指定Celery应用，
-l info
设置日志级别为info。
现在，你可以在你的代码中调用这个异步任务：
from my_project.tasks import add result = add.delay(4, 4) # 异步调用，返回AsyncResult对象 print(result.id) # 打印任务ID，可以用来追踪任务状态 # 稍后获取任务结果 # from celery.result import AsyncResult # result = AsyncResult(task_id='你的任务ID', app=app) # print(result.ready()) # 检查任务是否完成 # print(result.get()) # 获取任务结果
就是这样！Celery会将
add
任务放入消息队列，worker会从队列中取出任务并执行，并将结果存储在Redis中。
Celery的适用场景有哪些？
Celery非常适合处理需要较长时间才能完成的任务，例如：
- 发送电子邮件
- 处理图像或视频
- 执行复杂的计算
- 定期执行的任务（例如，每天凌晨备份数据库）
- 与外部API交互
简单来说，任何你不希望阻塞用户请求的操作，都可以交给Celery来处理。
如何监控Celery任务的执行情况？
监控Celery任务的执行情况至关重要，可以帮助你及时发现并解决问题。Celery本身并没有提供内置的监控工具，但你可以使用一些第三方工具，比如：
-
Flower: 一个基于Web的Celery监控工具，可以实时查看任务状态、worker状态、队列长度等信息。安装很简单：
pip install flower
，然后运行
celery -A my_project flower
。
-
Celery Beat: 用于调度定期任务，可以配置任务的执行时间、频率等。你需要创建一个配置文件，指定要执行的任务及其执行时间。
-
自定义监控: 你也可以自己编写监控脚本，通过Celery的API获取任务状态，并将数据存储到数据库或监控系统中。
选择哪种监控方式取决于你的具体需求和技术栈。Flower是一个不错的入门选择，简单易用，功能也比较完善。
如何处理Celery任务执行失败的情况？
任务执行失败是不可避免的，Celery提供了一些机制来处理这种情况：
- 重试: 你可以使用
retry
方法让Celery自动重试失败的任务。可以设置重试次数和重试间隔。
@shared_task(bind=True, max_retries=3) def add(self, x, y): try: # 模拟可能出错的操作 result = x / (y - 2) return result except Exception as exc: self.retry(exc=exc, countdown=5) # 5秒后重试
- 错误处理: 你可以使用
on_failure
方法来处理任务执行失败的情况。可以发送错误通知、记录日志等。
@shared_task(on_failure=error_handler) def my_task(x, y): return x + y def error_handler(uuid, args, kwargs, einfo): print(f"任务 {uuid} 执行失败: {einfo}") # 发送错误通知
- 死信队列: 将执行失败的任务放入死信队列，稍后进行分析和处理。
选择哪种处理方式取决于你的具体需求。对于一些临时性的错误，重试可能是一个不错的选择。对于一些无法自动恢复的错误，需要进行人工干预。
评论（已关闭）
评论已关闭