前言
无论是kafka，还是RocketMq，消费者方法参数中的MessageExt对象不能被 fastjson默认的方式序列化。
一、查看代码
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,ConsumeConcurrentlyContext context) {
try {
GiftSendMessage message = JSON.parseObject(msgs.get(0).getBody(),
GiftSendMessage.class);
UserInfo userInfo = new UserInfo();
userInfo.setMemberId(String.valueOf(message.getFromId()));
userInfo.setScId(message.getScId());
userInfo.setAnchorScid(message.getAnchorScid());
userInfo.setAnchorId(message.getAnchorId());
userInfo.setGiftSendToId(message.getToId());
DrawByGiftSendConfig draw =
filter.getGiftIdBySend().get(String.valueOf(message.getGiftId()));
GiftSendRequest request = new GiftSendRequest();
request.setActivityId(draw.getActivityId());
request.setBoxType(draw.getBoxType());
request.setGiftOrderId(message.getGiftOrderId());
request.setBatch(message.getAmount());
request.setSendTime(message.getSendTime());
request.setUserInfo(userInfo);
request.setPaymentVersion(PAYMENT_VERSION);
this.userDrawService.giftSend(request);
LOG.info("DRAW BY GIFT SEND WITH SANTA CHECKOUT, MQ-MESSAGE={}, REQUEST={}",
JSON.toJSONString(message),
JSON.toJSONString(request));
} catch (Exception e) {
LOG.error("GIFT-SEND-QUEUE CONSUME FATAL ERROR ON PARSING, DROPPED, MSG={},EXCEPTION={}, EXCEPTION-MESSAGE={}",
JSON.toJSONString(msgs), e.getClass().getSimpleName(),e.getMessage(), e);
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
二、问题代码
三、原因分析
上图中红色方框当中，用的序列化方式为 fastjson，此行代码会抛出异常，导致消费失败，进入重试队列，且没有任何业务日志输出。MQ源码如下：
如果异常，返回 ConsumeConcurrentlyStatus.RECONSUME_LATER;
结论：无论是kafka，还是RocketMq，消费者方法参数中的MessageExt对象不能被 fastjson默认的方式序列化。
环境：项目采用1.2.31 (最新版本1.2.78)
接下来，我们分析下fastjson序列化的完整过程：
fastjson反序列化的方式默认为采用 get方法、is方法作为序列化属性 字段的,序列化流程如下：
其中：在获取对象序列化的时候，MessageExt中有返回 ByteBuffer的get方法，代码如下：
public ByteBuffer getStoreHostBytes() {
return socketAddress2ByteBuffer(this.storeHost);
}
//socketAddress2ByteBuffer
public static ByteBuffer socketAddress2ByteBuffer(SocketAddress socketAddress) {
ByteBuffer byteBuffer = ByteBuffer.allocate(8);
return socketAddress2ByteBuffer(socketAddress, byteBuffer);
}
//socketAddress2ByteBuffer
public static ByteBuffer socketAddress2ByteBuffer(SocketAddress socketAddress, ByteBuffer
byteBuffer) {
InetSocketAddress inetSocketAddress = (InetSocketAddress)socketAddress;
byteBuffer.put(inetSocketAddress.getAddress().getAddress(), 0, 4);
byteBuffer.putInt(inetSocketAddress.getPort());
byteBuffer.flip();
return byteBuffer;
}
Mq消息在接收到消息时，构造了返回了ByteBuffer对象的方法，该方法是nio中设计用于保存数据到缓冲区的目的。
3.1 ByteBuffer主要的属性，与fastjson的get方法反序列化方式
- position: 其实是指从buffer读取或写入buffer的下一个元素位置。比如，已经写入buffer 3个元素那那么position就是指向第4个位置，即position设置为3（数组从0开始计）；
- limit：还有多少数据需要从buffer中取出，或还有多少空间可以放入；postition总是<=limit;
- capacity: 表示buffer本身底层数组的容量。limit绝不能>capacity；
数据结构如下：
- get()方法，一字节一字节读；
- getChar()、getShort()、getInt()、getFloat()、getLong()、getDouble()读取相应字节数的数据；
3.2 最终原因分析得到
至此：问题显而易见，fastjson在1.2.31及之前，没有提供ByteBuffer 序列化器，所以用了默认的javabean序列化器，而默认的javabean序列化器，又通过get方法反序列化，当遇见ByteBuffer时，ByteBuffer中会先遇到如下方法，getLong():
public long getLong() {
return Bits.getLong(this, ix(nextGetIndex(8)), bigEndian);
}
//nextGetIndex
final int nextGetIndex(int nb) { // package-private
if (limit - position < nb)
throw new BufferUnderflowException();
int p = position;
position += nb;
return p;
}
每次读取position偏移8个字节，而MessageExt中，构建的ByteBuffer存储的时4个字节，所以会报错，完整的堆栈如下：
四、总结
- fastjson 序列化需注意问题，不能序列化RocketMq工具类中MessageExt对象，会报错；
- 原因：MessageExt包含ByteBuffer类型的nio包对象，而低版本fastjson没有该类型序列化器（JavaBeanSerializer），使用默认的序列化器；
- ByteBuffer中getLong方法，getLong方法的position偏移8个字节，而MessageExt中，构建的ByteBuffer存储的时4个字节，所以会BufferUnderflowException异常。
- BufferUnderflowException异常：表示在读取缓冲区时发生了下溢（underflow）的情况。下溢是指尝试从缓冲区中读取比可用数据量更多的数据；
- 无论是kafka，还是RocketMq，消费者方法参数中的MessageExt对象不能被 fastjson默
认的方式序列化；
