分布式任务队列Celery的实践

╰+攻爆jí腚メ 2023-09-29 16:22 0阅读 0赞

笔者写下此文总结对 Celery 的了解和在工作中的使用。本文的大概内容如下：

1.  任务队列是什么；
2.  Celery 做了什么；
3.  Celery 在工作中的实践。

## 任务队列是什么 ##

“消息队列（Message Queue）”，后端同学应该都有了解，常见的有 RabbitMQ、RocketMQ、Kafka。而“任务队列（Task Queue）”，笔者在接触 Celery 之前是没有听过的。任务队列是什么，而任务队列和消息队列，这两者之间有何关系。带着问题，先看看 Celery 的架构：

![465e647d1405b23fe1562066438c3b93.png][]

在 Celery 的架构中，可看出由多台 Server 发起异步任务（Async Task），发送任务到 Broker 的队列中，其中的 Celery Beat 进程可负责发起定时任务。当 Task 到达 Broker 后，会将其分发给相应的 Celery Worker 进行处理。当 Task 处理完成后，其结果存储至 Backend。

在上述过程中的 Broker 和 Backend，Celery 没有实现，而是使用了现有开源实现，例如 RabbitMQ 作为 Broker 提供消息队列服务，Redis 作为 Backend 提供结果存储服务。Celery 就像是抽象了消息队列架构中 Producer、Consumer 的实现，将消息队列中基本单位“消息”抽象成了任务队列中的“任务”，并将异步、定时任务的发起和结果存储等操作进行了封装，让开发者可以忽略 AMQP、RabbitMQ 等实现细节，为开发带来便利。

综上所述，**Celery 作为任务队列是基于消息队列的进一步封装，其实现依赖消息队列**。

接下来，通过一个简单的应用来具体了解 Celery 做了什么。

## Celery 做了什么 ##

在应用开发中，为了保证响应速度，耗时且不影响流程的操作通常被做异步处理。例如在用户注册的处理过程中，通常会异步发送邮件通知用户，下面看看 Celery 是如何实现该异步操作。

在 task.py 中声明了发送邮件的方法 send\_mail，并为其加上 Celery 提供的 @app.task 装饰器。通过该装饰器，可以将 send\_mail 函数变成一个 celery.app.task:Task 实例对象。而该 **Task 实例可提供了两个核心功能**：

1.  **将消息发送给队列；**
2.  **声明 Worker 接收到消息后需要执行的具体函数。**

from celery import Celery
    
    app = Celery('tasks', broker='am

[465e647d1405b23fe1562066438c3b93.png]: https://img-blog.csdnimg.cn/img_convert/465e647d1405b23fe1562066438c3b93.png