RabbitMQ消息体含额外字符致JSON反序列化异常求助

问题现象 #

消费RabbitMQ队列消息时，反序列化为实体对象偶尔触发JSON格式异常：

• 异常1：Unterminated string. Expected delimiter: ". Path '***', line 1, position 1197.
• 异常2：Additional text encountered after finished reading JSON content: e. Path '', line 1, position 863.

反序列化前记录的消息体包含无关内容（如*2、$4、AUTH及乱码），示例如下：

*2
$4
AUTH
$18
insCGTICPassR3d50!
*3
$6
CLIENT
$7
SETNAME
$8
GW-WEB01
*3
$6
databases
*2
$8
SENTINEL
$7
MASTERS
*1
$4
INFO
*2
$7
CLUSTER
$5
NODES
*2
$4
ECHO
$16
����6�G��aV�\B
d":"CtUAAB+LCAAAAAAABADtXe+O2ji7//5K5x6iHumVjlYNCST8mbfLyhAD2fxtEmiZL4idYbecMjACptvZG9g72E+rXsPcw0zv69gJCWGIYzAeSvcgddTYj/3ws/38Hj92nOTN4gJOP40ms9uR8PlmMl1cLH589WG5vL0oFBZXH0Y3w4WI8hez4a04m/9WwBeF0apG4VX9zeKiMbu+r79pj5bubDK+ugd3yw+j6XJ8NVyOZ1NjdO+NFrez6WKlP9G+HN3c3s3HodZXlPp3k+UK3XAL3fVwObyaTZfz4dUyVFaUJKUgVQoQKVne+7fDq9FChJ+Xo/l0OBGjHxG90fJuPh1Pfwvub0evVsrHifLff/9d/L0Uq5ML7y3TD3/v9Xi6WA6nVyMEeXgBrhDSoT2rq2pFKkpvCqksLJ6M5kt0GQlWiTg/GH1eCuOL6Xjy46vl/G70qhCKUOtn8/EfYeMXYVZjuJgMb4bBcD7+UC9KxeJrW

涉事代码片段 #

消息发布代码 #

var message = JsonConvert.Serialize<T>(object);
var body = Encoding.UTF8.GetBytes(message);

var properties = Channel.CreateBasicProperties();
properties.Persistent = isPersistent;

if (headers != null)
  properties.Headers = headers;

Channel.BasicPublish(
   exchange: exchange,
   routingKey: routingKey,
   basicProperties: properties,
   body: body);

消息消费代码 #

var body = e.Body.ToArray();
message = Encoding.UTF8.GetString(body);
var requestModel = JsonConvert.DeserializeObject<T>(message);

action.Invoke(requestModel);

// notifies RabbitMq that the notification is complete.
((EventingBasicConsumer)sender).Model.BasicAck(deliveryTag: e.DeliveryTag, multiple: false);

问题分析 #

消息体中的*2、$4、AUTH等内容符合Redis RESP协议格式，说明错误的消息（Redis命令/响应）被误发送到了RabbitMQ队列。可能的原因：

• 服务配置错误，将Redis操作的输出定向到了RabbitMQ队列
• RabbitMQ的Channel对象被污染，比如与Redis客户端复用了连接资源
• 消息发布逻辑存在漏洞，误将其他服务的二进制/文本内容作为JSON消息发布
• 请求响应的压缩逻辑异常，导致二进制乱码混入JSON内容

解决方案 #

1. 定位错误消息来源：通过RabbitMQ控制台查看异常消息的生产者连接信息，定位发送错误消息的服务实例
2. 发布端校验：在序列化后添加JSON合法性校验，避免非法内容进入队列：

   var message = JsonConvert.Serialize<T>(object);
   // 预校验JSON格式
   try
   {
       JToken.Parse(message);
   }
   catch (JsonException ex)
   {
       // 记录日志并丢弃非法消息
       _logger.LogError(ex, "Failed to validate serialized JSON. Content: {Message}", message);
       return;
   }
   var body = Encoding.UTF8.GetBytes(message);
   // 后续发布逻辑...
   

3. 消费端容错：反序列化前先校验消息合法性，非法消息直接拒绝：

   var body = e.Body.ToArray();
   message = Encoding.UTF8.GetString(body);
   try
   {
       JToken.Parse(message);
   }
   catch (JsonException ex)
   {
       _logger.LogError(ex, "Invalid JSON message received. Body: {Message}", message);
       // 拒绝消息，避免重复消费
       ((EventingBasicConsumer)sender).Model.BasicNack(e.DeliveryTag, false, false);
       return;
   }
   var requestModel = JsonConvert.DeserializeObject<T>(message);
   // 后续消费逻辑...
   

4. 隔离资源：确保RabbitMQ客户端与Redis等其他服务的客户端完全隔离，禁止复用连接或通道对象
5. 检查压缩逻辑：如果请求响应采用压缩存储，需确保压缩后内容通过Base64编码转为文本再放入JSON，避免二进制数据破坏JSON结构

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