Python Celery多队列配置与路由实战指南
发布时间：2026-10-13| 4人浏览过
本文详细介绍Python Celery框架中多队列的配置方法。通过实例代码展示如何定义Queue对象、设置CELERY_QUEUES与CELERY_ROUTES参数，实现任务到指定队列和routing_key的精准路由。涵盖项目结构、配置文件编写及Worker启动命令，帮助开发者优化异步任务分发策略。
本文通过详细的代码示例，解析Python Celery框架中多队列的配置方法，旨在为开发者的学习或实际工作提供具有参考价值的技术方案。
参考Celery官方文档以获取更深入的配置说明。
项目结构与初始化配置
首先明确项目的目录结构，确保文件层级清晰。
/proj
-__init__
-app.py #实例化celery对象
-celeryconfig.py #celery的配置文件
-tasks.py #celery编写任务文件核心应用文件app.py负责初始化Celery实例。
#coding:utf-8
from __future__ import absolute_import
from celery import Celery
app = Celery('proj', include=['proj.tasks']) #实例化celery对象
app.config_from_object('proj.celeryconfig') #引入配置文件
if __name__ == '__main__':
app.start()- 其中proj参数指定了Celery实例的唯一名称。
- include参数用于指定启动时需自动导入的任务模块列表。
tasks.py文件用于编写具体的异步任务逻辑。
#coding:utf-8
from __future__ import absolute_import
from proj.app import app
@app.task()
def add(x, y):
return x + y队列与路由参数详解
celeryconfig.py文件集中管理Celery的各项配置参数。
#coding:utf-8
from kombu import Queue
BROKER_URL = 'amqp://guest:guest@127.0.0.1:5672//' # 使用RabbitMQ作为消息代理
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/0' # 把任务结果存在了Redis
CELERY_TASK_SERIALIZER = 'msgpack' # 任务序列化和反序列化使用msgpack方案
CELERY_RESULT_SERIALIZER = 'json' # 读取任务结果一般性能要求不高，所以使用了可读性更好的JSON
CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24 # 任务过期时间，不建议直接写86400，应该让这样的magic数字表述更明显
CELERY_ACCEPT_CONTENT = ['json', 'msgpack'] # 指定接受的内容类型
CELERY_QUEUES = ( #设置add队列,绑定routing_key
Queue('add', routing_key='xue.add'),
)
CELERY_ROUTES = { #projq.tasks.add这个任务进去add队列并routeing_key为xue.add
'projq.tasks.add': {
'queue': 'add',
'routing_key': 'xue.add',
}
}- CELERY_ACCEPT_CONTENT设置为msgpack，这是一种相比JSON体积更小、解析更快的序列化格式，使用此格式需安装对应的Python包。
- CELERY_QUEUES用于定义指定routing_key的队列，队列名称可根据实际需求任意指定。
- CELERY_ROUTES用于配置任务路由，将特定任务名映射到对应的队列及routing_key，需注意这里的routing_key必须与CELERY_QUEUES中定义的一致。
Worker启动与任务发布
接下来介绍如何启动Celery Worker进程。
在项目根目录（proj的上层目录）执行启动命令。
celery -A proj.app worker -Q add -l info命令中的proj.tasks.add对应CELERY_ROUTES中配置的任务名称。
参数-Q add指定监听的队列，key=xue.add对应配置的routing_key。
介绍如何发布异步任务至指定队列。
from proj.tasks import add
add.delay(2,3)多队列配置示例
在多队列模式下，配置文件需要做出相应调整。
CELERY_QUEUES = ( #设置add队列,绑定routing_key
Queue('add', routing_key='xue.add'),
)
CELERY_ROUTES = { #projq.tasks.add这个任务进去add队列并routeing_key为xue.add
'projq.tasks.add': {
'queue': 'add',
'routing_key': 'xue.add',
}示例中配置了两个独立的队列以处理不同任务。
# 配置队列
CELERY_QUEUES = (
Queue('default', routing_key='default'),
Queue('队列1', routing_key='key1'),
Queue('队列2', routing_key='key2'),
)
# 路由（哪个任务放入哪个队列）
CELERY_ROUTES = {
'任务1': {'queue': '队列1', 'routing_key': 'key1'},
'任务2': {'queue': '对列2', 'routing_key': 'key2'},
}以上内容涵盖了Celery多队列配置的核心要点，希望对读者有所帮助。
免责声明：本文内容仅供参考，不代表本站观点。转载或引用请注明出处。
