Django+Celery：CELERY_TASK_ROUTES模块相对路径与通配符失效问题

我来帮你排查下这个Celery路由不生效的问题，结合你给出的配置细节，咱们一步步定位并解决：

1. 先修正路由配置里的占位符错误 #

你当前的路由配置里写了{'queue': "..."}，这明显是个未替换的占位符，Celery根本不知道要把任务路由到哪个实际队列！把这个占位符改成你定义好的队列名（比如'my-queue'）是第一步：

CELERY_TASK_ROUTES = {
    # 'my-celery-task': {'queue': "my-queue"},
    'my_app.celery_tasks.*': {'queue': "my-queue"}
}

2. 检查Exchange类型与路由键的匹配 #

你的队列配置里用了routing_key='my-queue.#'这种带通配符的格式，但默认的Exchange类型是direct——而direct类型的Exchange只支持精确匹配路由键，#和*这类通配符只有topic类型的Exchange才支持！这会导致路由规则完全不生效，必须修改Exchange的类型：

from kombu import Queue, Exchange

CELERY_TASK_QUEUES = (
    Queue('default', Exchange('default', type='topic'), routing_key='default.#'),
    Queue('my-queue', Exchange('my-queue', type='topic'), routing_key='my-queue.#'),
)

3. 确保Celery加载了正确的路由规则 #

修改配置后，必须重启Celery Worker——Worker只会在启动时加载配置，热修改不会生效。重启后可以用以下命令验证当前Worker的路由配置：

celery -A your_django_project_name inspect routes

这个命令会输出Worker当前生效的所有路由规则，你可以直接看到my_app.celery_tasks.*是否被正确映射到my-queue。

4. 验证任务是否被正确发送到目标队列 #

你可以在Django Shell里手动触发一个任务，然后检查队列状态：

# 进入Django shell
python manage.py shell

# 导入任务并触发
from my_app.celery_tasks import your_test_task
your_test_task.delay()

之后可以用celery -A your_django_project_name inspect active_queues查看Worker监听的队列，或者直接查看Redis里的队列数据（比如用redis-cli执行LRANGE celery:queue:my-queue 0 -1），确认任务是否进入了my-queue。

额外提示：Celery 4.x的配置命名规范 #

虽然Celery 4.1.0还兼容旧的大写配置项（比如CELERY_TASK_ROUTES），但官方推荐使用小写的新规范（比如task_routes），你可以尝试切换成新规范，避免潜在的兼容性问题：

task_routes = {
    'my_app.celery_tasks.*': {'queue': "my-queue"}
}

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