分布式任务队列 Celery 的实践

发布时间：2025-11-02 21:12:12 来源：创站工坊 作者：应用开发

笔者在近期工作中有接触到 Celery，分布这是式任实践一个开源的分布式任务队列(Distributed Task Queue)，在 Github 上现有 18k star，分布主要可以用于实现应用中的式任实践异步任务和定时任务，虽然是分布用 Python 编写，但协议可以用任何语言实现，式任实践现已有 gocelery、分布nodecelery 和 celery-php 等。式任实践

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

任务队列是分布什么

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

Celery

在 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 实例可提供了两个核心功能：

Task 已经定义完成，若要发起异步任务，可通过调用 Task 的 delay 方法，该方法会将消息发送至队列，例如在用户注册完成时，发起发邮件的异步任务：

运行以上程序后，企商汇消息已经发送至 RabbitMQ 的队列中，可观察到其消息格式如下：

Task in RabbitMQ

可看出 Celery 封装后的消息包含了 task 标识和运行参数等内容。

接着，启动 Worker 消费 RabbitMQ 中的消息：

Worker 启动后，可以看到下面打印信息：

Worker Start

首先是 Worker 的配置信息，然后是 Worker 所执行的 Task 列表，接着是从 RabbitMQ 中成功获取消息并执行相应的 Task。

通过以上示例，可以进一步明白 Celery 作为任务队列框架所做的工作，而“分布式任务队列”中的”分布式“指的则是 Producer、Consumer 可以有多个，即多个进程向 Broker 发送任务，多个 Worker 从 Broker 中获取 Task 并执行。

以上只是一个简单的示例，接着再看下笔者在工作中所接触到的关于 Celery 使用的一些实践经验。

Celery 在工作中的实践

根据业务场景划分队列

在笔者所工作的项目中，Celery 用于处理下单、解析轨迹、推送上游等异步任务和定时任务。根据每个 Task 的业务场景，可为其指定对应的队列，例如：

根据业务场景，在 DEFAULT_CELERY_ROUTES 配置中指定 6 个 Task 对应的 Queue，共有 3 个队列 create、pull、push，并将该路由规则加入到 CELERY_ROUTES 中以生效。这样设计的目的是为了不同场景彼此之间互不影响，例如解析任务阻塞不应该影响下单任务。

进一步划分队列

在根据业务场景粗略划分后，对于某个场景，可能需要更细致的划分，例如在向上游推送时，为了避免一个上游的阻塞影响向其他上游推送，需要做到不同上游彼此之间互不影响。所以需要针对不同上游使用不同队列，例如：

在 CLIENT_CELERY_ROUTES 中指定了需要根据 Client 隔离队列的 Task 和其对应的 Queue 名称格式，队列名中含有一个占位符，为的是根据不同 Client 得到不同的队列名。

接着实现了一个路由器 ClientRouter ，其中定义了 router_for_task 方法，其作用是为 task 指定对应的队列名。可看出其中的逻辑是如果 task 在 CLIENT_CELERY_ROUTES 中，将会用 kwargs 中的 client_id 格式化队列名，得到最终发送消息的队列名，达到根据入参 client_id 来决定具体使用的队列，从而起到隔离不同 Client 使用不同队列的效果。

除了在 Client 的维度上划分，若需要在其他维度进一步划分队列以达到隔离的效果，也可参考该方法来设计路由规则。

动态队列

再来说说动态队列，其本质是预备队列，其目的是为了在线上环境减轻某些队列消息堆积的压力，起到快速支援的作用。通过配置来定义动态队列需要支援哪些队列，例如当 push 队列的压力较大，可配置 json 如下，将 push_tracking 和 push_weight 两个 Task 路由到预备的动态队列中。

上述配置的作用是将 70% 的 celery_task.push_tracking Task 路由到动态队列 1、2 上，70% 的 celery_task.push_weight Task 路由到动态队列 3、4 上。

动态队列的路由器 DynamicRouter 大致实现如下：

动态配置的定时任务

前文提到 Celery 不仅能实现异步任务，还能通过 Celery Beat 实现定时任务，首先看一个例子：

完成上述配置后，执行 Celery Beat 命令：

celery beat

即根据配置每 30 秒执行一次 send_email 任务。

上述示例是在代码中配置定时任务。而在笔者的工作中使用了 djcelery 提供的数据库调度模型，通过结合 django 提供的 ORM 功能来动态设置，更为方便。下面叙述如何实现，首先在 Celery 配置中新增：

设置使用 DatabaseScheduler，然后再生成定时任务的配置表：

可以看到数据库中多出了以下表：

完成以上操作，最后只用执行 Celery Beat 命令，则会去数据库中读取配置发起定时任务。这样的好处是可以通过修改数据库中的记录来实现动态配置定时任务，例如调整任务的周期或者参数。

以上便是笔者在工作中接触到 Celery 所收获的内容，如果有需要实现异步任务、定时任务的场景，可以考虑使用 Celery。

我是草捏子，一只热爱技术和生活的草鱼，我们下期见!

参考

Message Queue vs Task Queue difference (https://newbedev.com/message-queue-vs-task-queue-difference)

高性能异步框架Celery入坑指南 (https://juejin.cn/post/6844903689103081480)

分布式任务队列 Celery—深入 Task (https://www.cnblogs.com/yhilkfan-fanguiju/p/10589779.html)

相关文章