自定义协议解决TCP粘包

2、解决办法

2.1、 消息定长 ，报文大小固定长度，不够空格补全，发送和接收方遵循相同的约定，这样即使粘包了通过接收方编程实现获取定长报文也能区分。

2.2、 包尾添加特殊分隔符 ，例如每条报文结束都添加回车换行符（例如FTP协议）或者指定特殊字符作为报文分隔符，接收方通过特殊分隔符切分报文区分。

2.3、 将消息分为消息头和消息体 ，消息头中包含表示信息的总长度（或者消息体长度）的字段

3、自定义协议，来实现TCP的粘包/拆包问题

3.0  自定义协议，开始标记

3.1  自定义协议的介绍

3.2  自定义协议的类的封装

3.3  自定义协议的编码器

3.4  自定义协议的解码器

4、协议相关的实现

4.1  协议的封装

[java] view plain copy

print ?

1. import java.util.Arrays;  
2. /** 
3.  * <pre> 
4.  * 自己定义的协议 
5.  *  数据包格式 
6.  * +——----——+——-----——+——----——+ 
7.  * |协议开始标志|  长度             |   数据       | 
8.  * +——----——+——-----——+——----——+ 
9.  * 1.协议开始标志head_data，为int类型的数据，16进制表示为0X76 
10.  * 2.传输数据的长度contentLength，int类型 
11.  * 3.要传输的数据 
12.  * </pre> 
13.  */  
14. public class SmartCarProtocol {  
15.     /** 
16.      * 消息的开头的信息标志 
17.      */  
18.     private int head_data = ConstantValue.HEAD_DATA;  
19.     /** 
20.      * 消息的长度 
21.      */  
22.     private int contentLength;  
23.     /** 
24.      * 消息的内容 
25.      */  
26.     private byte[] content;  
27.     /** 
28.      * 用于初始化，SmartCarProtocol 
29.      *  
30.      * @param contentLength 
31.      *            协议里面，消息数据的长度 
32.      * @param content 
33.      *            协议里面，消息的数据 
34.      */  
35.     public SmartCarProtocol(int contentLength, byte[] content) {  
36.         this.contentLength = contentLength;  
37.         this.content = content;  
38.     }  
39.     public int getHead_data() {  
40.         return head_data;  
41.     }  
42.     public int getContentLength() {  
43.         return contentLength;  
44.     }  
45.     public void setContentLength(int contentLength) {  
46.         this.contentLength = contentLength;  
47.     }  
48.     public byte[] getContent() {  
49.         return content;  
50.     }  
51.     public void setContent(byte[] content) {  
52.         this.content = content;  
53.     }  
54.     @Override  
55.     public String toString() {  
56.         return "SmartCarProtocol [head_data=" + head_data + ", contentLength="  
57.                 + contentLength + ", content=" + Arrays.toString(content) + "]";  
58.     }  
59. }  

4.2  协议的编码器

[java] view plain copy

print ?

1. /** 
2.  * <pre> 
3.  * 自己定义的协议 
4.  *  数据包格式 
5.  * +——----——+——-----——+——----——+ 
6.  * |协议开始标志|  长度             |   数据       | 
7.  * +——----——+——-----——+——----——+ 
8.  * 1.协议开始标志head_data，为int类型的数据，16进制表示为0X76 
9.  * 2.传输数据的长度contentLength，int类型 
10.  * 3.要传输的数据 
11.  * </pre> 
12.  */  
13. public class SmartCarEncoder extends MessageToByteEncoder<SmartCarProtocol> {  
14.     @Override  
15.     protected void encode(ChannelHandlerContext tcx, SmartCarProtocol msg,  
16.             ByteBuf out) throws Exception {  
17.         // 写入消息SmartCar的具体内容  
18.         // 1.写入消息的开头的信息标志(int类型)  
19.         out.writeInt(msg.getHead_data());  
20.         // 2.写入消息的长度(int 类型)  
21.         out.writeInt(msg.getContentLength());  
22.         // 3.写入消息的内容(byte[]类型)  
23.         out.writeBytes(msg.getContent());  
24.     }  
25. }  

4.3  协议的解码器

[java] view plain copy

print ?

1. import java.util.List;  
2. import io.netty.buffer.ByteBuf;  
3. import io.netty.channel.ChannelHandlerContext;  
4. import io.netty.handler.codec.ByteToMessageDecoder;  
5. /** 
6.  * <pre> 
7.  * 自己定义的协议 
8.  *  数据包格式 
9.  * +——----——+——-----——+——----——+ 
10.  * |协议开始标志|  长度             |   数据       | 
11.  * +——----——+——-----——+——----——+ 
12.  * 1.协议开始标志head_data，为int类型的数据，16进制表示为0X76 
13.  * 2.传输数据的长度contentLength，int类型 
14.  * 3.要传输的数据,长度不应该超过2048，防止socket流的攻击 
15.  * </pre> 
16.  */  
17. public class SmartCarDecoder extends ByteToMessageDecoder {  
18.     /** 
19.      * <pre> 
20.      * 协议开始的标准head_data，int类型，占据4个字节. 
21.      * 表示数据的长度contentLength，int类型，占据4个字节. 
22.      * </pre> 
23.      */  
24.     public final int BASE_LENGTH = 4 + 4;  
25.     @Override  
26.     protected void decode(ChannelHandlerContext ctx, ByteBuf buffer,  
27.             List<Object> out) throws Exception {  
28.         // 可读长度必须大于基本长度  
29.         if (buffer.readableBytes() >= BASE_LENGTH) {  
30.             // 防止socket字节流攻击  
31.             // 防止，客户端传来的数据过大  
32.             // 因为，太大的数据，是不合理的  
33.             if (buffer.readableBytes() > 2048) {  
34.                 buffer.skipBytes(buffer.readableBytes());  
35.             }  
36.             // 记录包头开始的index  
37.             int beginReader;  
38.             while (true) {  
39.                 // 获取包头开始的index  
40.                 beginReader = buffer.readerIndex();  
41.                 // 标记包头开始的index  
42.                 buffer.markReaderIndex();  
43.                 // 读到了协议的开始标志，结束while循环  
44.                 if (buffer.readInt() == ConstantValue.HEAD_DATA) {  
45.                     break;  
46.                 }  
47.                 // 未读到包头，略过一个字节  
48.                 // 每次略过，一个字节，去读取，包头信息的开始标记  
49.                 buffer.resetReaderIndex();  
50.                 buffer.readByte();  
51.                 // 当略过，一个字节之后，  
52.                 // 数据包的长度，又变得不满足  
53.                 // 此时，应该结束。等待后面的数据到达  
54.                 if (buffer.readableBytes() < BASE_LENGTH) {  
55.                     return;  
56.                 }  
57.             }  
58.             // 消息的长度  
59.             int length = buffer.readInt();  
60.             // 判断请求数据包数据是否到齐  
61.             if (buffer.readableBytes() < length) {  
62.                 // 还原读指针  
63.                 buffer.readerIndex(beginReader);  
64.                 return;  
65.             }  
66.             // 读取data数据  
67.             byte[] data = new byte[length];  
68.             buffer.readBytes(data);  
69.             SmartCarProtocol protocol = new SmartCarProtocol(data.length, data);  
70.             out.add(protocol);  
71.         }  
72.     }  
73. }  

4.4  服务端加入协议的编/解码器

4.5  客户端加入协议的编/解码器

5、服务端的实现

[java] view plain copy

print ?

1. import io.netty.bootstrap.ServerBootstrap;  
2. import io.netty.channel.ChannelFuture;  
3. import io.netty.channel.ChannelInitializer;  
4. import io.netty.channel.ChannelOption;  
5. import io.netty.channel.EventLoopGroup;  
6. import io.netty.channel.nio.NioEventLoopGroup;  
7. import io.netty.channel.socket.SocketChannel;  
8. import io.netty.channel.socket.nio.NioServerSocketChannel;  
9. import io.netty.handler.logging.LogLevel;  
10. import io.netty.handler.logging.LoggingHandler;  
11. public class Server {  
12.     public Server() {  
13.     }  
14.     public void bind(int port) throws Exception {  
15.         // 配置NIO线程组  
16.         EventLoopGroup bossGroup = new NioEventLoopGroup();  
17.         EventLoopGroup workerGroup = new NioEventLoopGroup();  
18.         try {  
19.             // 服务器辅助启动类配置  
20.             ServerBootstrap b = new ServerBootstrap();  
21.             b.group(bossGroup, workerGroup)  
22.                     .channel(NioServerSocketChannel.class)  
23.                     .handler(new LoggingHandler(LogLevel.INFO))  
24.                     .childHandler(new ChildChannelHandler())//  
25.                     .option(ChannelOption.SO_BACKLOG, 1024) // 设置tcp缓冲区 // (5)  
26.                     .childOption(ChannelOption.SO_KEEPALIVE, true); // (6)  
27.             // 绑定端口 同步等待绑定成功  
28.             ChannelFuture f = b.bind(port).sync(); // (7)  
29.             // 等到服务端监听端口关闭  
30.             f.channel().closeFuture().sync();  
31.         } finally {  
32.             // 优雅释放线程资源  
33.             workerGroup.shutdownGracefully();  
34.             bossGroup.shutdownGracefully();  
35.         }  
36.     }  
37.     /** 
38.      * 网络事件处理器 
39.      */  
40.     private class ChildChannelHandler extends ChannelInitializer<SocketChannel> {  
41.         @Override  
42.         protected void initChannel(SocketChannel ch) throws Exception {  
43.             // 添加自定义协议的编解码工具  
44.             ch.pipeline().addLast(new SmartCarEncoder());  
45.             ch.pipeline().addLast(new SmartCarDecoder());  
46.             // 处理网络IO  
47.             ch.pipeline().addLast(new ServerHandler());  
48.         }  
49.     }  
50.     public static void main(String[] args) throws Exception {  
51.         new Server().bind(9999);  
52.     }  
53. }  

6、服务端Handler的实现

[java] view plain copy

print ?

1. import io.netty.channel.ChannelHandlerAdapter;  
2. import io.netty.channel.ChannelHandlerContext;  
3. public class ServerHandler extends ChannelHandlerAdapter {  
4.     // 用于获取客户端发送的信息  
5.     @Override  
6.     public void channelRead(ChannelHandlerContext ctx, Object msg)  
7.             throws Exception {  
8.         // 用于获取客户端发来的数据信息  
9.         SmartCarProtocol body = (SmartCarProtocol) msg;  
10.         System.out.println("Server接受的客户端的信息 :" + body.toString());  
11.         // 会写数据给客户端  
12.         String str = "Hi I am Server ...";  
13.         SmartCarProtocol response = new SmartCarProtocol(str.getBytes().length,  
14.                 str.getBytes());  
15.         // 当服务端完成写操作后，关闭与客户端的连接  
16.         ctx.writeAndFlush(response);  
17.         // .addListener(ChannelFutureListener.CLOSE);  
18.         // 当有写操作时，不需要手动释放msg的引用  
19.         // 当只有读操作时，才需要手动释放msg的引用  
20.     }  
21.     @Override  
22.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)  
23.             throws Exception {  
24.         // cause.printStackTrace();  
25.         ctx.close();  
26.     }  
27. }  

7、客户端的实现

[java] view plain copy

print ?

1. import io.netty.bootstrap.Bootstrap;  
2. import io.netty.channel.ChannelFuture;  
3. import io.netty.channel.ChannelInitializer;  
4. import io.netty.channel.ChannelOption;  
5. import io.netty.channel.EventLoopGroup;  
6. import io.netty.channel.nio.NioEventLoopGroup;  
7. import io.netty.channel.socket.SocketChannel;  
8. import io.netty.channel.socket.nio.NioSocketChannel;  
9. public class Client {  
10.     /** 
11.      * 连接服务器 
12.      *  
13.      * @param port 
14.      * @param host 
15.      * @throws Exception 
16.      */  
17.     public void connect(int port, String host) throws Exception {  
18.         // 配置客户端NIO线程组  
19.         EventLoopGroup group = new NioEventLoopGroup();  
20.         try {  
21.             // 客户端辅助启动类 对客户端配置  
22.             Bootstrap b = new Bootstrap();  
23.             b.group(group)//  
24.                     .channel(NioSocketChannel.class)//  
25.                     .option(ChannelOption.TCP_NODELAY, true)//  
26.                     .handler(new MyChannelHandler());//  
27.             // 异步链接服务器 同步等待链接成功  
28.             ChannelFuture f = b.connect(host, port).sync();  
29.             // 等待链接关闭  
30.             f.channel().closeFuture().sync();  
31.         } finally {  
32.             group.shutdownGracefully();  
33.             System.out.println("客户端优雅的释放了线程资源...");  
34.         }  
35.     }  
36.     /** 
37.      * 网络事件处理器 
38.      */  
39.     private class MyChannelHandler extends ChannelInitializer<SocketChannel> {  
40.         @Override  
41.         protected void initChannel(SocketChannel ch) throws Exception {  
42.             // 添加自定义协议的编解码工具  
43.             ch.pipeline().addLast(new SmartCarEncoder());  
44.             ch.pipeline().addLast(new SmartCarDecoder());  
45.             // 处理网络IO  
46.             ch.pipeline().addLast(new ClientHandler());  
47.         }  
48.     }  
49.     public static void main(String[] args) throws Exception {  
50.         new Client().connect(9999, "127.0.0.1");  
51.     }  
52. }  

8、客户端Handler的实现

[java] view plain copy

print ?

1. import io.netty.channel.ChannelHandlerAdapter;  
2. import io.netty.channel.ChannelHandlerContext;  
3. import io.netty.util.ReferenceCountUtil;  
4. //用于读取客户端发来的信息  
5. public class ClientHandler extends ChannelHandlerAdapter {  
6.     // 客户端与服务端，连接成功的售后  
7.     @Override  
8.     public void channelActive(ChannelHandlerContext ctx) throws Exception {  
9.         // 发送SmartCar协议的消息  
10.         // 要发送的信息  
11.         String data = "I am client ...";  
12.         // 获得要发送信息的字节数组  
13.         byte[] content = data.getBytes();  
14.         // 要发送信息的长度  
15.         int contentLength = content.length;  
16.         SmartCarProtocol protocol = new SmartCarProtocol(contentLength, content);  
17.         ctx.writeAndFlush(protocol);  
18.     }  
19.     // 只是读数据，没有写数据的话  
20.     // 需要自己手动的释放的消息  
21.     @Override  
22.     public void channelRead(ChannelHandlerContext ctx, Object msg)  
23.             throws Exception {  
24.         try {  
25.             // 用于获取客户端发来的数据信息  
26.             SmartCarProtocol body = (SmartCarProtocol) msg;  
27.             System.out.println("Client接受的客户端的信息 :" + body.toString());  
28.         } finally {  
29.             ReferenceCountUtil.release(msg);  
30.         }  
31.     }  
32.     @Override  
33.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)  
34.             throws Exception {  
35.         ctx.close();  
36.     }