如何在分离的Celery API与Worker服务中实现任务路由？

方法1：直接在send_task中指定队列参数 #

你误以为send_task不支持queue参数是个误解——它其实原生支持该参数。调用时直接传入queue就能把任务定向到指定队列：

celery_client.send_task('some_task', args=(some_arguments,), queue='queue_1')

执行后任务会被发送到queue_1，由监听该队列的Worker消费；要发往queue_2，只需把参数改成queue='queue_2'即可。

方法2：通过配置路由规则（适合批量任务路由） #

如果需要给某一类任务固定分配队列，也可以在API端的Celery客户端配置路由规则，完全不需要知晓任务的具体定义：

1. 初始化Celery客户端时配置task_routes：

from celery import Celery

celery_client = Celery(
    'api_client',
    broker='你的Broker地址（如Redis/RabbitMQ）',
    task_routes={
        'task_group_1.*': {'queue': 'queue_1'},  # 所有以task_group_1开头的任务自动发往queue_1
        'task_group_2.*': {'queue': 'queue_2'},  # 所有以task_group_2开头的任务自动发往queue_2
    }
)

2. 后续调用send_task时，只要任务名匹配规则，就会自动路由到对应队列：

celery_client.send_task('task_group_1.some_task', args=(some_arguments,))  # 自动进入queue_1

验证Worker配置 #

你的Worker启动命令是正确的，只需确保监听的队列名和API端指定的完全一致：

celery -A <path_to_my_celery_file>.celery_client worker --loglevel=info -Q queue_1
celery -A <path_to_my_celery_file>.celery_client worker --loglevel=info -Q queue_2

关键注意事项 #

• API端和Worker端的Celery客户端必须使用相同的Broker地址，否则无法完成任务通信。
• 如果用RabbitMQ作为Broker，还能通过指定exchange和routing_key实现更复杂的路由，但对你的场景来说，直接指定队列或简单路由规则已经足够满足需求。

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