使用Python从RabbitMQ取消息存MongoDB时MongoEngine实例化阻塞问题

1. 数据库连接别名不匹配 #

你的MongoDB连接使用了自定义别名raw-keyword-result，但TrackingRawKeyword文档类未指定使用该别名，MongoEngine会默认尝试调用default别名的连接（而该连接并未创建），导致实例化Document时卡在连接等待环节。

修复代码：
修改TrackingRawKeyword类，添加db_alias元数据：

class TrackingRawKeyword(Document):
    marketplaceId = StringField(required=True)
    market = StringField(choices=marketplaces, max_length=20, required=True)
    keyword = StringField(max_length=1024, required=True)
    ranking = IntField(required=True)
    date = DateTimeField()
    
    meta = {'db_alias': 'raw-keyword-result'}  # 指定使用自定义连接别名

2. 必填字段缺失 #

TrackingRawKeyword中的market字段标记为必填，但在handle_keyword_result实例化对象时，你仅传入了marketplace参数，未提供market字段值。这会触发ValidationError，但由于result_handler中的空except语句吞掉了所有异常，导致进程无提示挂起。

修复代码：
补充market字段的有效值（需匹配marketplaces选项）：

entry = TrackingRawKeyword(
    ranking=ranking,
    date=datetime.now(),
    marketplace=message['header']['marketplace'],
    market=message['header']['marketplace'],  # 补充必填的market字段
    keyword=message['data']['keyword'],
    marketplaceId=result['marketId']
)

3. 线程安全与连接校验优化 #

RabbitMQ的消费者回调在子线程中执行，虽然MongoEngine 0.10+支持多线程，但建议在回调中显式校验连接状态，避免因连接失效导致的挂起：

from mongoengine import get_connection, connect
from mongoengine.connection import ConnectionFailure

def handle_keyword_result(message):
    # 校验指定别名的连接是否可用
    try:
        get_connection(alias='raw-keyword-result')
    except ConnectionFailure:
        # 连接失效时重新建立
        connect(host=MONGO_HOST, alias='raw-keyword-result')
    
    # 原有业务代码...

4. 移除静默异常捕获 #

result_handler中的空except会吞掉所有异常（包括连接错误、数据验证错误等），导致无法定位问题。建议改为捕获特定异常并打印错误信息：

def result_handler(ch, method, properties, body):
    data = json.loads(body)
    job_type = data['header']['jobType']
    try:
        if job_type == 'DetailResult':
            handle_detail_result(data)
        elif job_type == 'KeywordResult':
            handle_keyword_result(data)
        print("yeyeyeye")
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception as e:
        print(f"消息处理失败: {str(e)}")
        # 异常时拒绝消息并重新入队（可选）
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)

调试步骤 #

1. 优先修复连接别名和必填字段问题
2. 恢复handle_keyword_result中的try-except并打印异常信息
3. 运行程序，根据错误提示进一步排查细节

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