问题:硬件传输数据被Netty分包了。
出现场景
硬件上报数据到系统时,数据包大小不是固定,有时对于超长的数据包,被Netty进行拆包发送了。
疑问
为什么说被Netty拆包了,因为TCP报文段的最大长度为65495字节,字节肯定没有超过这个数。
原因
Netty分配的缓冲区大小不是固定的。虽然Channel被创建时回去指定的缓冲区大小默认为 1024 。
单netty中有一个非常特别的机制。通过AdaptiveRecvByteBufAllocator类的doc文档可以看到。大致意思是:在反馈时自动增加或减少预测缓冲区大小。如果之前读取完全填满了分配的缓冲区,它会逐渐增加预期的可读字节数。如果读操作不能连续两次填充分配的缓冲区的一定数量,则会逐渐减少预期的可读字节数。
该操作应该是Netty节省内存开销设计的机制,非常有参考价值。有兴趣可以看看设计原理。在AbstractNioByteChannel 的read() 方法。通过debug 红框中开始进行缓冲区大小分配。最终算法(二分)在 AdaptiveRecvByteBufAllocator的getSizeTableIndex()方法中。
Netty 中对于拆包粘包的解决方案
- 使用netty提供固定的接收数组空间分配器来解决(最简单,但浪费的netty这个减少内存开销机制了)。把默认的AdaptiveRecvByteBufAllocator 改成 FixedRecvBufAllocator 。这样netty的缓冲区就是固定的1024大小了
- 代码进行合包(最优):数据包被进行拆包后,但数据包的顺序是不会改变的比如:
ASDC12345 ASDC123 + 45 数据包ASDC12345被拆成两段,但在传输过程中一定会是一前一后相邻的,顺序是不会改变的。这样我们就可以通过代码来进行粘包。为了安全还可以进行粘包后的数据包进行验证(需要协议包中加入校验码,或者约定在每个包中加入当前发送的数据包的大小)
private static final ConcurrentHashMap<String, String> PRESTORE_MAP = new ConcurrentHashMap<>();
/**
* 合包
*
* @param initialData 当前原始数据
* @param remoteAddress 通道远端地址
* @return 合包结果
*/
private String jointMsg(String initialData, String remoteAddress) {
String fixedHead = initialData.substring(CommonUtils.NumberUtil.NUM_ZERO, CommonUtils.NumberUtil.NUM_FOUR);
String fixedEnd = initialData.substring(initialData.length() - CommonUtils.NumberUtil.NUM_TWO);
String correctData = "";
if (!LimsProtocolConstant.HEAD.equals(fixedHead) && LimsProtocolConstant.END.equals(fixedEnd)) {
//原始数据中 无头有尾
String reserveValue = PRESTORE_MAP.get(remoteAddress);
if (!StringUtils.isEmpty(reserveValue)) {
String lenField = reserveValue.substring(CommonUtils.NumberUtil.NUM_FOUR, CommonUtils.NumberUtil.NUM_EIGHT);
int correctLen = DataTypeConvert.hex2int(lenField);
correctData = reserveValue + initialData;
int correctDataLen = correctData.length();
if (correctDataLen / CommonUtils.NumberUtil.NUM_TWO == correctLen) {
PRESTORE_MAP.remove(remoteAddress);
return correctData;
}
}
}
if (LimsProtocolConstant.HEAD.equals(fixedHead) && !LimsProtocolConstant.END.equals(fixedEnd)) {
//原始数据中 有头没尾
PRESTORE_MAP.put(remoteAddress, initialData);
}
if (!LimsProtocolConstant.HEAD.equals(fixedHead) && !LimsProtocolConstant.END.equals(fixedEnd)) {
//原始数据中,无头无尾
log.error("原始协议数据错误:{}", initialData);
throw new BizException("原始数据不符合协议,协议头和协议尾不符合约定!");
}
if (LimsProtocolConstant.HEAD.equals(fixedHead) && LimsProtocolConstant.END.equals(fixedEnd)) {
//原始数据中,有头有尾
correctData = initialData;
}
return correctData;
}