当前位置：首页 > article >正文

【能用】springboot集成netty，解码器处理数据过长的问题

article 2025/1/7 4:02:07

netty解码器处理数据过长的问题

处理数据过长的详细流程

当第一次 decode 调用时，如果数据不完整，decode 方法会直接返回，Netty 会保留 ByteBuf 中的数据。后续数据到达时，会再次调用 decode 方法，ByteBuf 会累积新到达的数据。
一旦 ByteBuf 中的数据足够完整解析，就会成功解析出 CustomMessage 并添加到 out 列表中，传递给后续的 YourBusinessLogicHandler 进行处理。
这样，通过检查数据的完整性并利用 Netty 的数据累积特性，就可以实现当数据过长时等待全部数据到达后再进行解码操作。你可以根据具体的协议和需求调整 CustomDataDecoder 中的长度检查逻辑，确保对不同长度的数据都能正确处理。
请注意，在实际应用中，可能需要考虑长时间等待数据不到达的情况，可以添加超时机制或其他错误处理机制，以避免资源的无限占用。例如，可以在 ChannelHandlerContext 上使用 channel().close() 关闭连接，或者使用 ctx.fireExceptionCaught() 抛出异常进行异常处理。

解决思路

明确协议格式

确定 dataLen 在数据中的具体位置。假设 dataLen 在 cmdLen 和 cmdPayload 之后，并且 cmdPayload 的长度是由 cmdLen 确定的。
需要先读取 cmdLen，然后读取 cmdPayload，接着根据 cmdPayload 的长度找到 dataLen 的位置。

逐步解析数据

按照协议规定的顺序和长度，逐步从 ByteBuf 中读取相应的数据。

示例代码（Java 和 Netty）：

public class TcpMessageDecoderHandler extends ByteToMessageDecoder {
    private static final Logger logger = LoggerFactory.getLogger(TcpHandleServiceImpl.class);

    @Override
    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf in, List<Object> out) throws Exception {
        try {
            // 检查是否有足够的字节读取 cmdLen
            if (in.readableBytes() < 2) {
                // 等待更多数据，因为 cmdLen 占 2 字节
                return;
            }

            // 读取 cmdLen（大端序）
            int cmdLen = in.readUnsignedShort();

            // 检查是否有足够的字节读取 cmdPayload
            if (in.readableBytes() < cmdLen) {
                // 数据不完整，重置读指针，等待更多数据
                in.resetReaderIndex();
                return;
            }

            // 读取 cmdPayload
            byte[] cmdPayload = new byte[cmdLen];
            in.readBytes(cmdPayload);
            String cmd = new String(cmdPayload, StandardCharsets.UTF_8);

            // 检查是否有足够的字节读取 dataLen
            if (in.readableBytes() < 2) {
                // 等待更多数据，因为 dataLen 占 2 字节
                in.resetReaderIndex();
                return;
            }

            // 读取 dataLen（大端序）
            int dataLen = in.readUnsignedShort();

            logger.info("可读字节:" + in.readableBytes());
            // 检查是否有足够的字节读取 dataPayload
            if (in.readableBytes() < dataLen) {
                // 数据不完整，重置读指针，等待更多数据
                in.resetReaderIndex();
                return;
            }

            // 读取 dataPayload
            byte[] dataPayload = new byte[dataLen];
            in.readBytes(dataPayload);

            // 在这里可以将 cmd 和 dataPayload 封装到自定义对象中，例如 CustomMessage
            TcpMessage tcpMessage = new TcpMessage(cmd, dataPayload);
            out.add(tcpMessage);
        } catch (Exception e){
            logger.error("读取字节流出错:" + e.getMessage());
            // 重置读指针
            in.resetReaderIndex();
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        logger.error("error:" + cause.getMessage());
    }
}

代码解释

首先检查 ByteBuf 中是否有足够的字节读取 cmdLen，如果不足 2 字节，等待更多数据。
读取 cmdLen，它是 2 字节的大端序数据。
检查是否有足够的字节读取 cmdPayload，如果不足，重置读指针等待更多数据。
读取 cmdPayload 并将其转换为字符串（假设是 UTF-8 编码）。
检查是否有足够的字节读取 dataLen，如果不足，重置读指针等待更多数据。
读取 dataLen，它是 2 字节的大端序数据。
检查是否有足够的字节读取 dataPayload，如果不足，重置读指针等待更多数据。
读取 dataPayload 并存储在字节数组中。
创建 CustomMessage 对象，将 cmd 和 dataPayload 作为成员变量，并将其添加到 out 列表中。