RabbitMQ.Client消费Azure Service Bus Shovel消息时抛出内部错误异常

问题场景 #

.NET应用使用RabbitMQ.Client 6.4.0连接RabbitMQ，通过Shovel将Azure Service Bus（ASB）主题portal.quoterequests的消息同步到RabbitMQ交换器portal.quoterequests。消费该交换器绑定队列的消息时触发以下异常，但直接通过RabbitMQ UI或RabbitMQ.Client发送的消息可正常消费：

RabbitMQ.Client.Exceptions.OperationInterruptedException: The AMQP operation was interrupted: AMQP close-reason, initiated by Peer, code=541, text='INTERNAL_ERROR', classId=0, methodId=0
at RabbitMQ.Client.Impl.SimpleBlockingRpcContinuation.GetReply(TimeSpan timeout)
at RabbitMQ.Client.Impl.ModelBase.BasicConsume(String queue, Boolean autoAck, String consumerTag, Boolean noLocal, Boolean exclusive, IDictionary2 arguments, IBasicConsumer consumer) at RabbitMQ.Client.Impl.AutorecoveringModel.BasicConsume(String queue, Boolean autoAck, String consumerTag, Boolean noLocal, Boolean exclusive, IDictionary2 arguments, IBasicConsumer consumer)
at RabbitMQ.Client.IModelExensions.BasicConsume(IModel model, String queue, Boolean autoAck, IBasicConsumer consumer)
at HiBunny.Program.Main(String[] args) in d:\study\HiBunny\HiBunny\Program.cs:line 64

核心原因 #

1. 消息元数据不兼容：ASB与RabbitMQ的消息属性（如自定义头、内容类型）格式或类型存在差异，Shovel未做适配，导致RabbitMQ解析时触发内部错误。
2. Shovel配置未做转换：默认Shovel仅转发消息体，未处理ASB特有的消息属性，这些属性在RabbitMQ中无法被正常解析。
3. 版本兼容性问题：RabbitMQ或Shovel插件版本与RabbitMQ.Client 6.4.0存在兼容性bug。

解决步骤 #

1. 配置Shovel的消息属性转换 #

修改Shovel配置，添加publish-properties参数映射ASB消息属性为RabbitMQ兼容格式：

shovel.asb-to-rabbit =
  src-uri = amqps://<你的ASB命名空间>.servicebus.windows.net/
  src-queue = portal.quoterequests
  dest-uri = amqp://localhost:5672/
  dest-exchange = portal.quoterequests
  # 强制设置内容类型，过滤ASB特有的属性
  publish-properties = content_type=application/json
  # 若需保留自定义属性，确保类型为字符串或基本类型
  publish-headers = x-asb-message-id=${headers.x-ms-message-id}

2. 消费端忽略异常属性 #

修改消费逻辑，跳过可能引发错误的消息属性，仅处理消息体：

public void HandleBasicDeliver(string consumerTag, ulong deliveryTag, bool redelivered, string exchange, string routingKey, IBasicProperties properties, ReadOnlyMemory<byte> body)
{
    try
    {
        // 直接解析消息体，避免处理可能有问题的属性
        var messageContent = Encoding.UTF8.GetString(body.Span);
        // 执行业务逻辑
        model.BasicAck(deliveryTag, false);
    }
    catch (Exception ex)
    {
        // 记录异常后拒绝消息，避免循环消费
        model.BasicNack(deliveryTag, false, false);
    }
}

3. 升级组件版本 #

• 将RabbitMQ升级至最新稳定版（如3.12.x），确保Shovel插件版本与RabbitMQ版本匹配。
• 升级RabbitMQ.Client至6.8.x及以上版本，修复已知的AMQP协议兼容性问题。

4. 启用RabbitMQ调试日志定位根因 #

修改rabbitmq.conf开启调试日志：

log.level = debug
log.file = /var/log/rabbitmq/debug.log

重启RabbitMQ后，查看日志中INTERNAL_ERROR的详细堆栈，可精准定位是消息属性、编码还是其他内部组件导致的错误。

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