Paho MQTT订阅者在RabbitMQ中消息堆积问题排查

你的理解确实存在偏差，出现google.protobuf.message.DecodeError时消息堆积，大概率是以下几个原因：

• 异常捕获范围不完整：如果protobuf解码逻辑没被包含在try-except块中，或者except语句未覆盖DecodeError（比如try块只包裹了业务处理代码，解码步骤在块外执行），抛出的异常会直接中断流程，导致ACK无法发送。

• ACK发送逻辑遗漏异常分支：如果你的代码仅在消息处理成功时发送ACK，异常场景下未执行ACK（或NACK）操作，RabbitMQ会判定这条消息未被处理完成，会一直留存队列甚至重复投递，最终造成堆积。必须确保无论处理成功还是失败（包括解码错误），都要显式发送ACK或NACK。

• RabbitMQ MQTT的ACK机制差异：RabbitMQ对MQTT的ACK处理有特殊逻辑，若订阅者使用QoS 1/2的手动确认模式，必须显式调用确认方法。如果依赖自动确认但客户端配置未生效，或者手动确认的调用时机不对，都会导致消息滞留。

修正后的逻辑示例 #

def on_message(client, userdata, message):
    try:
        # 把解码逻辑纳入try块
        proto_msg = YourProtoClass.FromString(message.payload)
        process_message(proto_msg)
        message.ack()
    except google.protobuf.message.DecodeError as e:
        # 解码错误时主动ACK，避免堆积
        message.ack()
        # 可选：记录错误日志，排查无效消息来源
        print(f"Protobuf解码失败: {e}, 消息payload: {message.payload[:32]}...")
    except Exception as e:
        # 其他异常同样执行确认操作
        message.ack()
        print(f"消息处理失败: {e}")

如果不想让解码错误的消息重复占用队列资源，也可以发送NACK并设置拒绝重发：message.nack(requeue=False)，这类消息会被直接丢弃或进入死信队列，不再堆积在原消费队列。

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