美文网首页netty
Netty之解决TCP粘包拆包(自定义协议)

Netty之解决TCP粘包拆包(自定义协议)

作者: 达微 | 来源:发表于2018-08-15 15:57 被阅读0次

    1、什么是粘包/拆包

    一般所谓的TCP粘包是在一次接收数据不能完全地体现一个完整的消息数据。TCP通讯为何存在粘包呢?主要原因是TCP是以流的方式来处理数据,再加上网络上MTU的往往小于在应用处理的消息数据,所以就会引发一次接收的数据无法满足消息的需要,导致粘包的存在。处理粘包的唯一方法就是制定应用层的数据通讯协议,通过协议来规范现有接收的数据是否满足消息数据的需要。

    2、解决办法

    2.1、消息定长,报文大小固定长度,不够空格补全,发送和接收方遵循相同的约定,这样即使粘包了通过接收方编程实现获取定长报文也能区分。
    2.2、包尾添加特殊分隔符,例如每条报文结束都添加回车换行符(例如FTP协议)或者指定特殊字符作为报文分隔符,接收方通过特殊分隔符切分报文区分。
    2.3、将消息分为消息头和消息体,消息头中包含表示信息的总长度(或者消息体长度)的字段3、自定义协议,来实现TCP的粘包/拆包问题

    3.0 自定义协议,开始标记

    3.1 自定义协议的介绍
    3.2 自定义协议的类的封装
    3.3 自定义协议的编码器
    3.4 自定义协议的解码器

    4、协议相关的实现

    4.1 协议的封装

    import java.util.Arrays; 
    /** *
    * 自己定义的协议
    
    *  数据包格式
    
    * +——----——+——-----——+——----——+
    
    * |协议开始标志|  长度            |  数据      |
    
    * +——----——+——-----——+——----——+
    
    * 1.协议开始标志head_data,为int类型的数据,16进制表示为0X76
    
    * 2.传输数据的长度contentLength,int类型
    
    * 3.要传输的数据
    *
    */public class SmartCarProtocol {
    /** * 消息的开头的信息标志
     */
    private int head_data = ConstantValue.HEAD_DATA;
    /** * 消息的长度 */
    private int contentLength;
    /** * 消息的内容 */
    private byte[] content;
     /** 
    * 用于初始化,SmartCarProtocol 
    * @param contentLength 
    *   协议里面,消息数据的长度
    * @param content
     *            协议里面,消息的数据 
    */
    public SmartCarProtocol(int contentLength, byte[] content)
     {
    this.contentLength = contentLength;this.content = content;}
     public int getHead_data()
     {return head_data;} public int getContentLength() {return contentLength;} 
    public void setContentLength(int contentLength) {this.contentLength = contentLength;}
     public byte[] getContent() {return content;} 
    public void setContent(byte[] content) {this.content = content;} 
    @Overridepublic String toString() {return "SmartCarProtocol [head_data=" + head_data + ", contentLength="+ contentLength + ", content=" + Arrays.toString(content) + "]";} }      
    

    4.2 协议的编码器

    /** *
    * 自己定义的协议
    
    *  数据包格式
    
    * +——----——+——-----——+——----——+
    
    * |协议开始标志|  长度            |  数据      |
    
    * +——----——+——-----——+——----——+
    
    * 1.协议开始标志head_data,为int类型的数据,16进制表示为0X76
    
    * 2.传输数据的长度contentLength,int类型
    
    * 3.要传输的数据
    
    *
    
    */public class SmartCarEncoder extends MessageToByteEncoder{ @Overrideprotected void encode(ChannelHandlerContext tcx, SmartCarProtocol msg,ByteBuf out) throws Exception {
    // 写入消息SmartCar的具体内容// 1.写入消息的开头的信息标志(int类型)out.writeInt(msg.getHead_data());
    // 2.写入消息的长度(int 类型)out.writeInt(msg.getContentLength());// 3.写入消息的内容(byte[]类型)out.writeBytes(msg.getContent());}}    
    

    4.3 协议的解码器

    import java.util.List;import io.netty.buffer.ByteBuf;import io.netty.channel.ChannelHandlerContext;import io.netty.handler.codec.ByteToMessageDecoder; /** *
    * 自己定义的协议
    
    *  数据包格式
    
    * +——----——+——-----——+——----——+
    
    * |协议开始标志|  长度            |  数据      |
    
    * +——----——+——-----——+——----——+
    
    * 1.协议开始标志head_data,为int类型的数据,16进制表示为0X76
    
    * 2.传输数据的长度contentLength,int类型
    
    * 3.要传输的数据,长度不应该超过2048,防止socket流的攻击
    
    *
    
    */public class SmartCarDecoder extends ByteToMessageDecoder { /** *
    
    * 协议开始的标准head_data,int类型,占据4个字节.
    
    * 表示数据的长度contentLength,int类型,占据4个字节.
    
    *
    
    */public final int BASE_LENGTH = 4 + 4; @Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf buffer,Listout) throws Exception {// 可读长度必须大于基本长度if (buffer.readableBytes() >= BASE_LENGTH) {// 防止socket字节流攻击// 防止,客户端传来的数据过大// 因为,太大的数据,是不合理的if (buffer.readableBytes() > 2048) {buffer.skipBytes(buffer.readableBytes());} // 记录包头开始的indexint beginReader; while (true) {// 获取包头开始的indexbeginReader = buffer.readerIndex();// 标记包头开始的indexbuffer.markReaderIndex();// 读到了协议的开始标志,结束while循环if (buffer.readInt() == ConstantValue.HEAD_DATA) {break;} // 未读到包头,略过一个字节// 每次略过,一个字节,去读取,包头信息的开始标记buffer.resetReaderIndex();buffer.readByte(); // 当略过,一个字节之后,// 数据包的长度,又变得不满足// 此时,应该结束。等待后面的数据到达if (buffer.readableBytes() < BASE_LENGTH) {return;}} // 消息的长度 int length = buffer.readInt();// 判断请求数据包数据是否到齐if (buffer.readableBytes() < length) {// 还原读指针buffer.readerIndex(beginReader);return;} // 读取data数据byte[] data = new byte[length];buffer.readBytes(data); SmartCarProtocol protocol = new SmartCarProtocol(data.length, data);out.add(protocol);}} }    
    

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

    5、服务端的实现

    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.logging.LogLevel;import io.netty.handler.logging.LoggingHandler; public class Server { public Server() {} public void bind(int port) throws Exception {// 配置NIO线程组EventLoopGroup bossGroup = new NioEventLoopGroup();EventLoopGroup workerGroup = new NioEventLoopGroup();try {// 服务器辅助启动类配置ServerBootstrap b = new ServerBootstrap();b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).handler(new LoggingHandler(LogLevel.INFO)).childHandler(new ChildChannelHandler())//.option(ChannelOption.SO_BACKLOG, 1024) // 设置tcp缓冲区 // (5).childOption(ChannelOption.SO_KEEPALIVE, true); // (6)// 绑定端口 同步等待绑定成功ChannelFuture f = b.bind(port).sync(); // (7)// 等到服务端监听端口关闭f.channel().closeFuture().sync();} finally {// 优雅释放线程资源workerGroup.shutdownGracefully();bossGroup.shutdownGracefully();}} /** * 网络事件处理器 */private class ChildChannelHandler extends ChannelInitializer{@Overrideprotected void initChannel(SocketChannel ch) throws Exception {// 添加自定义协议的编解码工具ch.pipeline().addLast(new SmartCarEncoder());ch.pipeline().addLast(new SmartCarDecoder());// 处理网络IOch.pipeline().addLast(new ServerHandler());}} public static void main(String[] args) throws Exception {new Server().bind(9999);}}6、服务端Handler的实现import io.netty.channel.ChannelHandlerAdapter;import io.netty.channel.ChannelHandlerContext; public class ServerHandler extends ChannelHandlerAdapter {// 用于获取客户端发送的信息@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg)throws Exception {// 用于获取客户端发来的数据信息SmartCarProtocol body = (SmartCarProtocol) msg;System.out.println("Server接受的客户端的信息 :" + body.toString()); // 会写数据给客户端String str = "Hi I am Server ...";SmartCarProtocol response = new SmartCarProtocol(str.getBytes().length,str.getBytes());// 当服务端完成写操作后,关闭与客户端的连接ctx.writeAndFlush(response);// .addListener(ChannelFutureListener.CLOSE); // 当有写操作时,不需要手动释放msg的引用// 当只有读操作时,才需要手动释放msg的引用} @Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)throws Exception {// cause.printStackTrace();ctx.close();}}7、客户端的实现import io.netty.bootstrap.Bootstrap;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.NioSocketChannel; public class Client { /** * 连接服务器 *  * @param port * @param host * @throws Exception */public void connect(int port, String host) throws Exception {// 配置客户端NIO线程组EventLoopGroup group = new NioEventLoopGroup();try {// 客户端辅助启动类 对客户端配置Bootstrap b = new Bootstrap();b.group(group)//.channel(NioSocketChannel.class)//.option(ChannelOption.TCP_NODELAY, true)//.handler(new MyChannelHandler());//// 异步链接服务器 同步等待链接成功ChannelFuture f = b.connect(host, port).sync(); // 等待链接关闭f.channel().closeFuture().sync(); } finally {group.shutdownGracefully();System.out.println("客户端优雅的释放了线程资源...");} } /** * 网络事件处理器 */private class MyChannelHandler extends ChannelInitializer {
    
    @Override
    
    protected void initChannel(SocketChannel ch) throws Exception {
    
    // 添加自定义协议的编解码工具
    
    ch.pipeline().addLast(new SmartCarEncoder());
    
    ch.pipeline().addLast(new SmartCarDecoder());
    
    // 处理网络IO
    
    ch.pipeline().addLast(new ClientHandler());
    
    }
    
    }
    
    public static void main(String[] args) throws Exception {
    
    new Client().connect(9999, "127.0.0.1");
    
    }
    
    }
    

    8、客户端Handler的实现

    import io.netty.channel.ChannelHandlerAdapter;
    
    import io.netty.channel.ChannelHandlerContext;
    
    import io.netty.util.ReferenceCountUtil;
    
    //用于读取客户端发来的信息
    
    public class ClientHandler extends ChannelHandlerAdapter {
    
    // 客户端与服务端,连接成功的售后
    
    @Override
    
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
    
    // 发送SmartCar协议的消息
    
    // 要发送的信息
    
    String data = "I am client ...";
    
    // 获得要发送信息的字节数组
    
    byte[] content = data.getBytes();
    
    // 要发送信息的长度
    
    int contentLength = content.length;
    
    SmartCarProtocol protocol = new SmartCarProtocol(contentLength, content);
    
    ctx.writeAndFlush(protocol);
    
    }
    
    // 只是读数据,没有写数据的话
    
    // 需要自己手动的释放的消息
    
    @Override
    
    public void channelRead(ChannelHandlerContext ctx, Object msg)
    
    throws Exception {
    
    try {
    
    // 用于获取客户端发来的数据信息
    
    SmartCarProtocol body = (SmartCarProtocol) msg;
    
    System.out.println("Client接受的客户端的信息 :" + body.toString());
    
    } finally {
    
    ReferenceCountUtil.release(msg);
    
    }
    
    }
    
    @Override
    
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
    
    throws Exception {
    
    ctx.close();
    
    }
    
    }
    

    9、参考的博客地址

    http://www.cnblogs.com/whthomas/p/netty-custom-protocol.html

    http://www.cnblogs.com/fanguangdexiaoyuer/p/6131042.html

    相关文章

      网友评论

        本文标题:Netty之解决TCP粘包拆包(自定义协议)

        本文链接:https://www.haomeiwen.com/subject/mdswbftx.html