Celery+MongoDB配置异常：任务始终写入kombu_default集合

核心问题梳理 #

你遇到的两个关键问题：任务始终写入默认的kombu_default集合；修改数据库名后生成了空库，本质是对Celery MongoDB Broker的配置逻辑不熟悉，以下是具体排查和解决步骤：

1. 任务落入默认集合的原因：Broker URL未指定目标集合 #

Celery使用MongoDB作为消息中间件时，默认将任务存在kombu_default集合。你的BROKER_URL只替换了数据库名，没有指定集合参数，所以无论配置多少队列，任务都会自动落入默认集合。

解决方法：

在MongoDB连接URL中添加collection参数，指定要写入的集合名：

# 先修改settings中的mongodb_host模板，加入collection占位符
# mongodb_host="mongodb+srv://{username}:{password}@cluster0.sq3xm.mongodb.net/{database}?collection={collection}&retryWrites=true&w=majority&appName=Cluster0"

# 再生成带集合参数的Broker URL
BROKER_URL = settings.mongodb_host.format(
    username=settings.mongodb_username,
    password=settings.mongodb_password,
    database=settings.cel_broker_db_name,
    collection=settings.cel_broker_collection_name  # 比如本地用"local_tasks"，生产用"prod_tasks"
)

2. task_queues/task_routes配置对MongoDB Broker无效 #

MongoDB Broker不支持RabbitMQ那样的exchange、queue绑定机制，你写的task_queues和task_routes配置在这里完全不起作用，属于无效配置，可以直接删除。

如果需要按任务类型分不同集合，要么给不同任务配置独立的Broker URL，要么通过broker_transport_options全局指定集合：

celery.conf.update(
    # ... 其他配置
    broker_transport_options={
        "database": settings.cel_broker_db_name,
        "collection": settings.cel_broker_collection_name,
    },
)

注：URL参数的优先级高于broker_transport_options，二选一即可。

3. 修改数据库名后无数据写入的原因 #

原因1：定时任务配置有重复键

你的beat_schedule里有两个同名的"execute_save_tweets_chain"键，Python字典不允许重复键，后面的配置会直接覆盖前面的，导致其中一个定时任务完全不触发。

修正方法：

给重复的键名加后缀区分：

celery.conf.beat_schedule = {
    "write-tweet-for-ai_nympho": {
        "task": "write_tweet_for_user",
        "schedule": 3600.0,
        "args": ("1866555368765100032",)
    }, 
    "write-tweet-for-hsingh": {
        "task": "write_tweet_for_user",
        "schedule": 3600.0,
        "args": ("1761349378227286016",)
    },
    "execute_save_tweets_chain-ai_nympho": {
        "task": "execute_save_tweets_chain",
        "schedule": 7200.0,
        "args": ("1866555368765100032",)
    },
    "execute_save_tweets_chain-hsingh": {
        "task": "execute_save_tweets_chain",
        "schedule": 7200.0,
        "args": ("1761349378227286016",)
    }
}

原因2：Worker未连接到新数据库

启动Worker时加上--loglevel=info，查看日志里输出的Broker URL是否正确加载了新数据库名，确认Worker和Beat都用了最新配置。

4. Backend配置冗余问题 #

你在BACKEND_URL里已经指定了数据库，又在backend_settings中重复指定database，会导致配置冲突。二选一即可：

# 方式1：通过URL指定集合
BACKEND_URL = settings.mongodb_host.format(
    username=settings.mongodb_username,
    password=settings.mongodb_password,
    database=settings.cel_backend_db_name
) + f"?taskmeta_collection={settings.cel_backend_collection_name}"

# 方式2：通过backend_settings配置
celery.conf.update(
    backend=f"mongodb+srv://{settings.mongodb_username}:{settings.mongodb_password}@cluster0.sq3xm.mongodb.net/",
    backend_settings={
        "database": settings.cel_backend_db_name,
        "taskmeta_collection": settings.cel_backend_collection_name
    },
    # ... 其他配置
)

验证步骤 #

1. 启动Worker：celery -A celery_app worker --loglevel=info，检查日志中的Broker URL是否包含正确的数据库和集合名。
2. 手动调用任务，查看MongoDB对应数据库的指定集合是否出现任务文档。
3. 启动Beat：celery -A celery_app beat --loglevel=info，确认四个定时任务都能正常触发。

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