Apache NMS IBytesMessage出队后带前缀致反序列化失败求助

针对你遇到的.NET NMS消费Node.js Rhea发送的AMQP消息时出现的IBytesMessage前缀问题，给出以下几个可行的解决方向：

1. 提取IBytesMessage中的有效JSON内容 #

先确定前缀的具体结构和长度，再截取有效部分进行反序列化：

• 在.NET消费代码中添加调试逻辑，输出消息字节的原始内容：

  if (message is IBytesMessage bytesMessage)
  {
      byte[] fullBytes = new byte[bytesMessage.BodyLength];
      bytesMessage.ReadBytes(fullBytes);
      
      // 输出十六进制格式便于分析前缀
      Console.WriteLine($"Raw bytes: {BitConverter.ToString(fullBytes)}");
      // 输出原始字符串
      Console.WriteLine($"Raw string: {Encoding.UTF8.GetString(fullBytes)}");
      
      // 根据实际输出调整前缀长度，示例为8字节，截取有效JSON部分
      byte[] jsonBytes = fullBytes.Skip(8).ToArray();
      string jsonContent = Encoding.UTF8.GetString(jsonBytes);
      var data = JsonConvert.DeserializeObject<YourModel>(jsonContent);
  }
  

• 确定前缀固定长度后，直接跳过前缀即可完成反序列化。

2. 调整Node.js Rhea的消息发送格式 #

让发送的AMQP消息能被ActiveMQ Classic正确转换为OpenWire的TextMessage，而非BytesMessage：

• 发送时明确指定消息的content_type为application/json或text/plain，确保消息体是纯JSON字符串：

  const rhea = require('rhea');
  const connection = rhea.connect({host: 'your-activemq-host', port: 5672});
  
  connection.on('sender_open', (context) => {
      const payload = {key: 'value', data: 123};
      const message = {
          body: JSON.stringify(payload),
          content_type: 'application/json',
          durable: true // 匹配.NET端的消费持久化设置
      };
      context.sender.send(message);
  });
  
  connection.open_sender('your-queue-name');
  

• 避免发送Buffer类型的消息体，直接传递字符串，减少ActiveMQ转换时的歧义。

3. 配置ActiveMQ Classic优化跨协议转换 #

修改ActiveMQ的activemq.xml配置，确保AMQP到OpenWire的消息转换正确：

• 在<transportConnectors>下的AMQP连接器中添加消息转换器配置：

  <transportConnector name="amqp" uri="amqp://0.0.0.0:5672?transport.transformer=org.apache.activemq.transport.amqp.AmqpMessageTransformer"/>
  

• 该转换器会自动处理AMQP消息到OpenWire消息的类型映射，将AMQP的TextMessage转换为OpenWire的TextMessage，这样.NET端就能直接获取ITextMessage，无需处理前缀。

4. 手动解析AMQP消息结构 #

如果上述方法无效，可以手动解析IBytesMessage中的AMQP消息帧，提取body内容：

• AMQP 0-9-1消息的帧结构以AMQP四个字节开头，后续是版本号和帧数据。参考AMQP规范，跳过帧头部分后提取消息体：

  byte[] fullBytes = /* 从IBytesMessage读取的字节数组 */;
  // 跳过AMQP帧头（前8字节："AMQP" + 版本0,0,9,1）
  int headerLength = 8;
  byte[] bodyBytes = fullBytes.Skip(headerLength).ToArray();
  // 进一步解析AMQP的消息体结构，提取JSON内容
  

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