如何配置Celery多CELERY_BROKER_URL实现指定任务路由分发

1. 修改环境变量配置 #

在原有环境配置文件中新增第二个Broker的地址参数，修改后内容如下：

CELERY_TASK_TRACK_STARTED=True
# 默认Broker地址
CELERY_BROKER_URL="redis://127.0.0.1:6379"
# 新增第二套独立Broker地址
CELERY_BROKER_URL_2="redis://127.0.0.2:6379"

2. 更新celery.py配置 #

替换原有celery.py内容，新增多Broker路由规则，核心是给不同Broker绑定独立队列、配置任务匹配规则：

import os
from celery import Celery
from kombu import Exchange, Queue

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'engine.settings')

app = Celery('engine')
app.config_from_object('django.conf:settings')

# 保留原有基础配置
app.conf.task_send_sent_event = True
app.conf.database_engine_options = {'echo': True}

# 多Broker路由核心配置
default_broker = os.getenv("CELERY_BROKER_URL", "redis://127.0.0.1:6379")
second_broker = os.getenv("CELERY_BROKER_URL_2", "redis://127.0.0.2:6379")

# 为两个Broker分别定义独立交换机
default_exchange = Exchange('default', type='direct')
second_exchange = Exchange('second_broker_ex', type='direct')

# 定义绑定对应Broker的任务队列
app.conf.task_queues = (
    Queue(
        'default',
        default_exchange,
        routing_key='default',
        broker=default_broker
    ),
    Queue(
        'second_broker_tasks',
        second_exchange,
        routing_key='second_broker_tasks',
        broker=second_broker
    ),
)

# 配置默认路由参数
app.conf.task_default_queue = 'default'
app.conf.task_default_exchange = 'default'
app.conf.task_default_routing_key = 'default'

# 配置任务路由规则：指定任务走第二个Broker
# 替换下面的任务路径为你实际需要转发的任务完整导入路径
app.conf.task_routes = {
    'your_app.tasks.specific_task': {
        'queue': 'second_broker_tasks',
        'routing_key': 'second_broker_tasks',
        'exchange': 'second_broker_ex'
    }
}

# 保留原有自动发现任务逻辑
app.autodiscover_tasks()

@app.task(bind=True)
def debug_task(self):
    print('Request: {0!r}'.format(self.request))

3. 启动Worker进程 #

两个Broker完全独立，需要分别启动Worker进程连接对应地址、监听专属队列：

• 启动默认Broker对应的Worker，处理常规任务：
  celery -A engine worker -Q default --broker=redis://127.0.0.1:6379 -l info
• 启动第二个Broker对应的Worker，处理指定路由的任务：
  celery -A engine worker -Q second_broker_tasks --broker=redis://127.0.0.2:6379 -l info

配置说明：如果有多个任务需要路由到第二个Broker，直接在task_routes字典中追加对应任务的路径规则即可，任务路径格式为所属Django App名.tasks.任务函数名。

内容的提问来源于stack exchange，提问作者Altafahmed