netty编解码器

当 Netty 发送或者接受一个消息的时候，就将会发生一次数据转换。入站消息会被解码：从字节转换为另一种格式（比如 java 对象）；如果是出站消息，它会被编码成字节。
Netty 提供一系列实用的编解码器，他们都实现了 ChannelInboundHadnler 或者 ChannelOutboundHandler 接口。在这些类中，channelRead 方法已经被重写了。以入站为例，对于每个从入站 Channel 读取的消息，这个方法会被调用。随后，它将调用由解码器所提供的 decode() 方法进行解码，并将已经解码的字节转发给 ChannelPipeline 中的下一个 ChannelInboundHandler。

解码器- ByteToMessageDecoder

关系继承图

小细节

不论解码器 handler 还是编码器 handler 即接收的消息类型必须与待处理的消息类型一致，否则该 handler 不会被执行。
当我们在handler中调用ctx.writeAndFlush()方法后，就会将数据交给ChannelOutboundHandler进行出站处理(如果下一个handler是编码器并且数据类型符合编码器处理的数据类型则会传递给编码器)，只是我们没有去定义出站类而已，若有需求可以自己去实现ChannelOutboundHandler出站类。
解码器一般不需要设置泛型(待处理数据的类型)。

TCP粘包和拆包基本介绍

1.TCP 是面向连接的，面向流的，提供高可靠性服务。收发两端（客户端和服务器端）都要有一一成对的 socket，因此，发送端为了将多个发给接收端的包，更有效的发给对方，使用了优化方法（Nagle 算法），将多次间隔较小且数据量小的数据，合并成一个大的数据块，然后进行封包。这样做虽然提高了效率，但是接收端就难于分辨出完整的数据包了，因为面向流的通信是无消息保护边界的
2.由于 TCP 无消息保护边界,需要在接收端处理消息边界问题，也就是我们所说的粘包、拆包问题

TCP粘包和拆包解决方案

具体事例：
1.要求客户端发送 5 个 Message 对象，客户端每次发送一个 Message 对象
2.服务器端每次接收一个 Message，分 5 次进行解码，每读取到一个 Message，会回复一个 Message 对象给客户端。
代码：
MessageProtocol.class

package com.haust.tcp;public class MessageProtocol {//协议包private int len; //关键private byte[] content;//内容public int getLen() {return len;}public void setLen(int len) {this.len = len;}public byte[] getContent() {return content;}public void setContent(byte[] content) {this.content = content;}
}

MyServer.class

package com.haust.tcp;import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;public class MyServer  {public static void main(String[] args) throws Exception{EventLoopGroup bossGroup = new NioEventLoopGroup(1);EventLoopGroup workerGroup = new NioEventLoopGroup();try {ServerBootstrap serverBootstrap = new ServerBootstrap();serverBootstrap.group(bossGroup,workerGroup).option(ChannelOption.SO_BACKLOG, 128).childOption(ChannelOption.SO_KEEPALIVE, true).channel(NioServerSocketChannel.class).childHandler(new MyServerInitializer()); //自定义一个初始化类ChannelFuture channelFuture = serverBootstrap.bind(7000).sync();channelFuture.channel().closeFuture().sync();}finally {bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}}

MyServerInitializer.class

package com.haust.tcp;import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;public class MyServerInitializer extends ChannelInitializer<SocketChannel> {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();pipeline.addLast(new MyMessageDecoder());//解码器pipeline.addLast(new MyMessageEncoder());//编码器pipeline.addLast(new MyServerHandler());}
}

MyServerHandler.class

package com.haust.tcp;import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;import java.nio.charset.Charset;
import java.util.Scanner;
import java.util.UUID;//处理业务的handler
public class MyServerHandler extends SimpleChannelInboundHandler<MessageProtocol> {private int count;@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {//cause.printStackTrace();ctx.close();}@Overrideprotected void channelRead0(ChannelHandlerContext ctx, MessageProtocol msg) throws Exception {//        接收到数据，并处理int len = msg.getLen();byte[] content = msg.getContent();System.out.println("服务器接收到信息如下");System.out.println("长度=" + len);System.out.println("内容=" + new String(content, Charset.forName("utf-8")));System.out.println("服务器接收到消息包数量=" + (++this.count));//回复消息System.out.println("服务端开始回复消息------");String responseContent = UUID.randomUUID().toString();int responseLen = responseContent.getBytes("utf-8").length;byte[]  responseContent2 = responseContent.getBytes("utf-8");//构建一个协议包MessageProtocol messageProtocol = new MessageProtocol();messageProtocol.setLen(responseLen);messageProtocol.setContent(responseContent2);//客户端需要输入信息，创建一个扫描器
//        MessageProtocol messageProtocol = new MessageProtocol();
//        int inLen=0;
//        byte[] incontent=null ;
//        Scanner scanner = new Scanner(System.in);
//        while (scanner.hasNextLine()) {//            String msg1 = scanner.nextLine();
//            //将字符串加工成协议包类型
//            inLen = msg1.getBytes().length;
//            incontent = msg1.getBytes();
//            messageProtocol.setLen(inLen);
//            messageProtocol.setContent(incontent);
//            //通过ctx 发送到服务器端ctx.writeAndFlush(messageProtocol);
//        }}
}

MyClient.class

package com.haust.tcp;import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;import java.util.Scanner;public class MyClient {public static void main(String[] args)  throws  Exception{EventLoopGroup group = new NioEventLoopGroup();try {Bootstrap bootstrap = new Bootstrap();bootstrap.group(group).channel(NioSocketChannel.class).handler(new MyClientInitializer()); //自定义一个初始化类ChannelFuture channelFuture = bootstrap.connect("localhost", 7000).sync();channelFuture.channel().closeFuture().sync();}finally {group.shutdownGracefully();}}
}

MyClientInitializer.class

package com.haust.tcp;import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;public class MyClientInitializer extends ChannelInitializer<SocketChannel> {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();pipeline.addLast(new MyMessageEncoder()); //加入编码器pipeline.addLast(new MyMessageDecoder()); //加入解码器pipeline.addLast(new MyClientHandler());}
}

MyClientHandler.class

package com.haust.tcp;import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;import java.nio.charset.Charset;public class MyClientHandler extends SimpleChannelInboundHandler<MessageProtocol> {private int count;@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {//使用客户端发送10条数据 "今天天气冷，吃火锅" 编号for(int i = 0; i< 5; i++) {String mes = "今天天气冷，吃火锅";byte[] content = mes.getBytes(Charset.forName("utf-8"));int length = mes.getBytes(Charset.forName("utf-8")).length;//创建协议包对象MessageProtocol messageProtocol = new MessageProtocol();messageProtocol.setLen(length);messageProtocol.setContent(content);ctx.writeAndFlush(messageProtocol);}}//    @Overrideprotected void channelRead0(ChannelHandlerContext ctx, MessageProtocol msg) throws Exception {int len = msg.getLen();byte[] content = msg.getContent();System.out.println("客户端接收到消息如下");System.out.println("长度=" + len);System.out.println("内容=" + new String(content, Charset.forName("utf-8")));System.out.println("客户端接收消息数量=" + (++this.count));}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {System.out.println("异常消息=" + cause.getMessage());ctx.close();}
}

MyMessageDecoder.class

package com.haust.tcp;import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ReplayingDecoder;import java.util.List;public class MyMessageDecoder extends ReplayingDecoder<Void> {@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {System.out.println();System.out.println();System.out.println("MyMessageDecoder decode 被调用");//需要将得到二进制字节码-> MessageProtocol 数据包(对象)int length = in.readInt();byte[] content = new byte[length];in.readBytes(content);//封装成 MessageProtocol 对象，放入 out， 传递下一个handler业务处理MessageProtocol messageProtocol = new MessageProtocol();messageProtocol.setLen(length);messageProtocol.setContent(content);//放入out传给下一个hanlder进行处理out.add(messageProtocol);}
}

MyMessageEncoder.class

package com.haust.tcp;import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;public class MyMessageEncoder extends MessageToByteEncoder<MessageProtocol> {@Overrideprotected void encode(ChannelHandlerContext ctx, MessageProtocol msg, ByteBuf out) throws Exception {System.out.println("MyMessageEncoder encode 方法被调用");out.writeInt(msg.getLen());out.writeBytes(msg.getContent());}
}

效果：
server端输出

MyMessageDecoder decode 被调用
服务器接收到信息如下
长度=27
内容=今天天气冷，吃火锅
服务器接收到消息包数量=1
服务端开始回复消息------
MyMessageEncoder encode 方法被调用MyMessageDecoder decode 被调用
服务器接收到信息如下
长度=27
内容=今天天气冷，吃火锅
服务器接收到消息包数量=2
服务端开始回复消息------
MyMessageEncoder encode 方法被调用MyMessageDecoder decode 被调用
服务器接收到信息如下
长度=27
内容=今天天气冷，吃火锅
服务器接收到消息包数量=3
服务端开始回复消息------
MyMessageEncoder encode 方法被调用MyMessageDecoder decode 被调用
服务器接收到信息如下
长度=27
内容=今天天气冷，吃火锅
服务器接收到消息包数量=4
服务端开始回复消息------
MyMessageEncoder encode 方法被调用MyMessageDecoder decode 被调用
服务器接收到信息如下
长度=27
内容=今天天气冷，吃火锅
服务器接收到消息包数量=5
服务端开始回复消息------
MyMessageEncoder encode 方法被调用

client端输出

MyMessageEncoder encode 方法被调用
MyMessageEncoder encode 方法被调用
MyMessageEncoder encode 方法被调用
MyMessageEncoder encode 方法被调用
MyMessageEncoder encode 方法被调用MyMessageDecoder decode 被调用
客户端接收到消息如下
长度=36
内容=f71b9a7d-96c0-4f26-a2ed-5684c30f0a5a
客户端接收消息数量=1MyMessageDecoder decode 被调用
客户端接收到消息如下
长度=36
内容=53cee25c-4a81-4d06-8ec8-bf6a664729e8
客户端接收消息数量=2MyMessageDecoder decode 被调用
客户端接收到消息如下
长度=36
内容=556d84ce-907b-4228-a722-9965b54e589e
客户端接收消息数量=3MyMessageDecoder decode 被调用
客户端接收到消息如下
长度=36
内容=774c3992-a39d-4aa2-9a20-e7fdbd3d4abd
客户端接收消息数量=4MyMessageDecoder decode 被调用
客户端接收到消息如下
长度=36
内容=69f8777f-dc61-4f2f-8f28-fdb0680c96d9
客户端接收消息数量=5