Netty-TCP服务端粘包、拆包问题（两种格式）

前言

最近公司搞了个小业务，需要使用TCP协议，我这边负责服务端。客户端是某个设备，客户端传参格式、包头包尾等都是固定的，不可改变，而且还有个蓝牙传感器，透传数据到这个设备，然后通过这个设备发送到服务端，数据格式也是不可变的。于是，相当于我这个TCP客户端会发送两种不同格式、不同长度的报文，且一种是ASCII 一种是HEX。

正常单发肯定是没问题的，但是，如果你业务卡顿，那么一定会有粘包、拆包的问题

请看：我在这里打个断点，模拟阻塞

然后一起发消息

放开断点

或者，睡个五秒

发现数据一起过来了，这就是粘包

还有种情况，如下

粘包了但是下一次的数据包部分字节出现在了上次的数据包的尾部，把整个数据包给分开了，这种就是拆包(大概就是整个效果)

总结就是：

粘包，就是将多个小的包封装成一个大的包进行发送。（多次发送的数据到了服务端合并成了一个数据包）

拆包，即是将一个超过缓冲区可用大小的包拆分成多个包进行发送。（一个的数据包到了服务端变得不完整了，哪怕是粘包都没有完整的一段）

那么如何解决呢？

本篇就以netty来简单说下，第一次用，很多不足，希望各位大佬指点！

直接上代码：

先来个maven:

<dependency><groupId>io.netty</groupId><artifactId>netty-all</artifactId><version>4.1.68.Final</version></dependency>

1. NettyTcpServerConfig TCP服务配置类

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import javax.annotation.PreDestroy;/*** @title: NettyTCPConfig* @description:* @date: 2024/10/14* @author: zwh* @copyright: Copyright (c) 2024* @version: 1.0*/
@Configuration
public class NettyTcpServerConfig {private final EventLoopGroup bossGroup = new NioEventLoopGroup();private final EventLoopGroup workerGroup = new NioEventLoopGroup();private ChannelFuture channelFuture;@Beanpublic ServerBootstrap serverBootstrap(NettyTcpServerHandler nettyTcpServerHandler) {ServerBootstrap bootstrap = new ServerBootstrap();bootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) {// 添加自定义解码器ch.pipeline().addLast(new MyCustomDecoder());// 自带的解码器   上面数据拆包分包就是用的这个自带的// ch.pipeline().addLast(new StringDecoder());ch.pipeline().addLast(new StringEncoder());ch.pipeline().addLast(nettyTcpServerHandler);}}).childOption(ChannelOption.TCP_NODELAY, true);return bootstrap;}public void startServer(int port) throws Exception {channelFuture = serverBootstrap(new NettyTcpServerHandler()).bind(port).sync();System.out.println("服务器已启动，监听端口: " + port);channelFuture.channel().closeFuture().sync();}@PreDestroypublic void shutdown() {workerGroup.shutdownGracefully();bossGroup.shutdownGracefully();}
}

2. NettyTcpServer TCP服务端启动入口

import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;/*** @title: NettyTcpServer* @description:* @date: 2024/10/14* @author: zwh* @copyright: Copyright (c) 2024* @version: 1.0*/@Component
public class NettyTcpServer implements CommandLineRunner {// 默认10067  可配置@Value("${nettyTcp.server.port:10067}")private int nettyTcpServerPort;private final NettyTcpServerConfig nettyTCPConfig;public NettyTcpServer(NettyTcpServerConfig nettyTcpServerConfig) {this.nettyTCPConfig = nettyTcpServerConfig;}@Overridepublic void run(String... args) throws Exception {nettyTCPConfig.startServer(nettyTcpServerPort);}
}

3. NettyTcpServerHandler 消息接收及回声(响应)处理

import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.springframework.stereotype.Component;import java.net.InetSocketAddress;/*** @title: NettyTcpServerHandler* @description:* @date: 2024/10/14* @author: zwh* @copyright: Copyright (c) 2024* @version: 1.0*//**   * @ChannelHandler.Sharable注解表示一个ChannelHandler实例可以被添加到多个ChannelPipeline中，并且该实例是线程安全的。‌ * 这意味着，如果一个ChannelHandler被标记为@Sharable，那么它可以在不同的ChannelPipeline中被共享使用，* 而不会出现竞争条件或线程安全问题。* */
@ChannelHandler.Sharable
@Component
public class NettyTcpServerHandler extends SimpleChannelInboundHandler<String> {@Overrideprotected void channelRead0(ChannelHandlerContext ctx, String msg) {System.out.println("接收到消息: " + msg);try {Thread.sleep(5000L);} catch (InterruptedException e) {throw new RuntimeException(e);}InetSocketAddress remoteAddress = (InetSocketAddress) ctx.channel().remoteAddress();String clientIp = remoteAddress.getAddress().getHostAddress();int clientPort = remoteAddress.getPort();System.out.println("来自客户端 (" + clientIp + ":" + clientPort + ") 的消息: " + msg);// 可根据需要发送响应String response = "Message processed: " + msg;ctx.writeAndFlush(response + "\r\n");
/*System.out.println("来自客户端 (" + clientIp + ":" + clientPort + ") 的消息: " + msg);if (msg.contains("重要")) {String responseMessage = "接收到重要数据: " + msg;ctx.writeAndFlush(responseMessage);System.out.println("发送响应到客户端 (" + clientIp + ":" + clientPort + "): " + responseMessage);} else {System.out.println("收到不重要数据，未发送响应。");}*/}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {cause.printStackTrace();ctx.close();}
}

4. MyCustomDecoder 自定义解码器

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import org.springframework.stereotype.Component;import java.nio.charset.StandardCharsets;
import java.util.List;/*** @title: MyMessageDecoder* @description:* @date: 2024/10/15* @author: zwh* @copyright: Copyright (c) 2024* @version: 1.0*/
@Component
public class MyCustomDecoder extends ByteToMessageDecoder{@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {while (in.isReadable()) {// ASCII 消息处理（第一个字节是$）if (in.getByte(in.readerIndex()) == '$') {// 查找结束符 '*'int endIndex = in.indexOf(in.readerIndex(), in.writerIndex(), (byte) '*');if (endIndex == -1) {// 还没有找到结束符，等待更多数据  这是有结束位的break;}// 读取完整的 ASCII 消息ByteBuf messageBuffer = in.readBytes(endIndex + 1 - in.readerIndex()); // 包括结束符String message = messageBuffer.toString(StandardCharsets.US_ASCII);out.add(message+decToHex(calculateBcc(message)));}// 十六进制消息处理（第一个字节是0x2B）else if (in.getByte(0) == (byte) 0x2B) {// 读取原始的十六进制数据// 创建一个新的 ByteBuf 来保存 15 个字节// 读取15个字节  另一种格式就是15个字节   然后读取后，原始的 ByteBuf 中的数据会被更新 将其标记为已读取下次就读不到了ByteBuf byteBuf = in.readBytes(15);// 将 ByteBuf 中的数据转换为十六进制字符串StringBuilder hexMessage = new StringBuilder();for (int i = 0; i < byteBuf.readableBytes(); i++) {byte b = byteBuf.getByte(i);hexMessage.append(String.format("%02X ", b));}// 将十六进制消息发送到下一个处理器out.add(hexMessage.toString().trim());// 释放 ByteBuf，避免内存泄漏byteBuf.release();}else {in.skipBytes(1); // 跳过无效字节}}}/*** 计算给定数据的 BCC 校验值** @param data 输入的字节数组* @return BCC 校验值*/public static byte calculateBcc(byte[] data) {byte bcc = 0;for (byte b : data) {bcc ^= b; // 使用异或运算计算 BCC}return bcc;}/*** 计算给定字符串中 $ 和 * 之间的 BCC 校验值** @param input 输入的字符串* @return BCC 校验值*/public static byte calculateBcc(String input) {byte bcc = 0;// 找到 $ 和 * 的位置int start = input.indexOf('$') + 1; // 从 $ 后开始int end = input.indexOf('*');// 确保找到 $ 和 * 的位置if (start > 0 && end > start) {String data = input.substring(start, end); // 提取 $ 和 * 之间的部分// 计算 BCCfor (char c : data.toCharArray()) {bcc ^= c; // 使用异或运算计算 BCC}}return bcc;}/*** 十进制转HEX(16进制)* @date 2024/10/22 15:22* @return {@link  }* @author zwh*/public static String decToHex(int dec) {return Integer.toHexString(dec).toUpperCase();}/*** 十六进制转HEX(10进制)* @date 2024/10/22 15:22* @return {@link  }* @author zwh*/public static Integer hexToDec(String hexValue) {return Integer.parseInt(hexValue, 16);}public static void main(String[] args) {decToHex(107);}
}

这个自定义解码器就看需要处理的数据类型了。我这里是两种数据：

某蓝牙传感器：$TB,6300,D702,4700,,84C2E4DCEAAD,*6B （$ 固定头 * 结束符，ASCII）

6B是 BCC异或校验，取 $ 和 * 之间所有值 (包括逗号)的异或校验参考BCC测试

I/O控制的器：2B 50 00 00 00 09 11 44 00 20 00 01 02 00 00 15个字节（HEX）

以这两种为示例来测试我们自定义的解码是否正确，如果需要别的数据自行修改开头和结尾以及长度啥的

Let's give it a try

就直接按照我们开头说的那种方法看看效果：