19.netty入门（十九）Netty入站与出站机制
19.netty入门（十九）Netty入站与出站机制
1.基本说明
（1）netty 的组件设计：Netty的主要组件有 Channel、EventLoop、ChannelFuture、ChannelHandler、ChannelPipe 等；
（2）ChannelHandler 充当了处理入站和出站数据的应用程序逻辑的容器。例如，实现 ChannelInboundHandler 接口（或 ChannelInboundHandlerAdapter），你就可以接收入站事件和数据，这些数据会被业务逻辑处理。当要给客户端发送响应时，也可以从 ChannelInboundHandler 冲刷数据。业务逻辑通常卸载一个或者多个 ChannelInboundHandler 中。ChannelOutboundHandler 原理一样，只不过它是用来处理出站数据的。
（3）ChannelPipeline 提供了 ChannelHandler 链的容器。以客户端应用程序为例，如果事件的运动方向是从客户端到服务端的，那么我们称这些事件为出站的，即客户端发送服务端的数据会通过 pipeline 中的一系列 ChannelOutboundHandler，并被这些 Handler 处理，反之则称为入站的
2.编码解码器
（1）当 Netty 发送或者接收一个消息的时候，就将会发生一次数据转换。入站消息会被解码：从字节转换为另一种格式（比如 java 对象）；如果是出站消息，它会被编码成字节。
（2）Netty 提供一系列实用的编解码器，他们都实现了 ChannelInboundHandler 或者 ChannelOutboundHandler 接口。在这些类中，channelRead 方法已经被重写了。已入站为例，对于每个从入站 Channel 读取的消息，这个方法会被调用。随后，它将调用由解码器所提供的 decode() 方法进行解码，并将已经解码的字节转发给 ChannelPipeline 中的下一个 ChannelInboundHandler。
2.1解码器 - ByteToMessageDecoder
（1）关系继承图
（2）由于不可能知道远程节点是否会一次性发送一个完成的信息，tcp有可能出现粘包拆包的问题，这个类会对入站数据进行缓冲，直到它准备好被处理。
（3）关于ByteToMessageDecoder实例分析
public class ToIntegerDecoder extends ByteToMessageDecoder {
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
if(in.readableBytes() >=4){
out.add(in.readInt());
}
}
}说明：
（1）这个例子，每次入站从 ByteBuf 读取4字节，将其解码为一个 int，然后将它添加到下一个 List 中。当没有更多元素可以被添加到该 List 中，它的内容将会发送给下一个 ChannelInboundHandler。int 在被添加到 List 中时，会被自动装箱为 Integer。在调用 readInt() 方法前必须验证所输入的 ByteBuf 是否具有足够的数据。
（2）decode执行分析图。
2.2Netty的handler链的调用机制
要求：
（1）使用自定义的编码器和解码器来说明 Netty 的 handler 调用机制。
2.2.1客户端发送给服务端
服务端代码
public class MyServer {
public static void main(String[] args) throws InterruptedException {
NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
NioEventLoopGroup workerGroup = new NioEventLoopGroup(8);
try {
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new MyServerInitializer());
ChannelFuture channelFuture = serverBootstrap.bind(7000).sync();
channelFuture.channel().closeFuture().sync();
}finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}public class MyServerInitializer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
// 入站的 handler 进行解码
pipeline.addLast(new MyByteToLongDecoder());
pipeline.addLast(new MyServerHandler());
}
}public class MyByteToLongDecoder extends ByteToMessageDecoder {
/**
* decode 会根据接收的数据，被调用多次，知道确定没有新的元素被添加到list，或者是Bytebuf 没有更多的可读字节为止
* 如果 list out 不为空，就会将list的内容传递给下一个 channelInboundHandler 处理，该处理器的方法也会被调用多次
*
* @param ctx 上下文
* @param in 入站的 ByteBuf
* @param out List 集合，将解码后的数据传给下一个 handler 处理
* @throws Exception
*/
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
System.out.println("MyByteToLongDecoder 的 decode()方法被调用");
// 因为 long 是 8 个字节，需要判断有 8 个字节，才能读取一个 long
if(in.readableBytes() >= 8){
out.add(in.readLong());
}
}
}public class MyServerHandler extends SimpleChannelInboundHandler<Long> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, Long msg) throws Exception {
System.out.println("从客户端" + ctx.channel().remoteAddress() + "读取到的数据到long " + msg);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}客户端代码
public class MyClinet {
public static void main(String[] args) throws InterruptedException {
NioEventLoopGroup 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();
}
}
}public class MyClientInitializer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
//加入一个出站的handler，对数据进行一个编码
pipeline.addLast(new MyLongToByteEncoder());
//加入一个自定义的handler，处理业务逻辑
pipeline.addLast(new MyClientHandler());
}
}public class MyLongToByteEncoder extends MessageToByteEncoder<Long> {
@Override
protected void encode(ChannelHandlerContext ctx, Long msg, ByteBuf out) throws Exception {
System.out.println("MyLongToByteEncoder 的 encode()方法被调用");
System.out.println("msg = " + msg);
out.writeLong(msg);
}
}public class MyClientHandler extends SimpleChannelInboundHandler<Long> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, Long msg) throws Exception {
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.out.println("MyClientHandler 发送数据");
ctx.writeAndFlush(123456L);
// 分析
// 1. "abcdabcdabcdabcd" 是16个字节
// 2. 该处理器的前一个 handler 是 MyLongToByteEncoder
// 3. MyLongToByteEncoder 父类是 MessageToByteEncoder
// 4. 父类 MessageToByteEncoder 中 acceptOutboundMessage(msg) 会判断当前 msg 是不是应该处理的类型，如果是就处理，不是就跳过
// 5. 编写 Encoder 时要注意传入的数据类型和处理的数据类型需要一致
// ctx.writeAndFlush(Unpooled.copiedBuffer("abcdabcdabcdabcd", CharsetUtil.UTF_8));
}
}2.2.2服务端发送给客户端
客户端增加一个Decoder，服务端增加一个Encoder。
修改服务端代码
public class MyServerInitializer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
// 入站的 handler 进行解码
pipeline.addLast(new MyByteToLongDecoder());
pipeline.addLast(new MyLongToByteEncoder());
pipeline.addLast(new MyServerHandler());
}
}public class MyServerHandler extends SimpleChannelInboundHandler<Long> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, Long msg) throws Exception {
System.out.println("从客户端" + ctx.channel().remoteAddress() + "读取到的数据到long " + msg);
// 给客户端发送一个 long
ctx.writeAndFlush(123L);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}修改客户端代码
public class MyClientInitializer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new MyLongToByteEncoder());
pipeline.addLast(new MyByteToLongDecoder());
pipeline.addLast(new MyClientHandler());
}
}public class MyClientHandler extends SimpleChannelInboundHandler<Long> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, Long msg) throws Exception {
System.out.println("服务器的ip=" + ctx.channel().remoteAddress());
System.out.println("服务器回送的消息：" + msg );
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.out.println("MyClientHandler 发送数据");
ctx.writeAndFlush(123456L);
// 分析
// 1. "abcdabcdabcdabcd" 是16个字节
// 2. 该处理器的前一个 handler 是 MyLongToByteEncoder
// 3. MyLongToByteEncoder 父类是 MessageToByteEncoder
// 4. 父类 MessageToByteEncoder 中 acceptOutboundMessage(msg) 会判断当前 msg 是不是应该处理的类型，如果是就处理，不是就跳过
// 5. 编写 Encoder 时要注意传入的数据类型和处理的数据类型需要一致
// ctx.writeAndFlush(Unpooled.copiedBuffer("abcdabcdabcdabcd", CharsetUtil.UTF_8));
}
}2.2.3ChannelPipeline类中的图
* <pre>
* I/O Request
* via {@link Channel} or
* {@link ChannelHandlerContext}
* |
* +---------------------------------------------------+---------------+
* | ChannelPipeline | |
* | \|/ |
* | +---------------------+ +-----------+----------+ |
* | | Inbound Handler N | | Outbound Handler 1 | |
* | +----------+----------+ +-----------+----------+ |
* | /|\ | |
* | | \|/ |
* | +----------+----------+ +-----------+----------+ |
* | | Inbound Handler N-1 | | Outbound Handler 2 | |
* | +----------+----------+ +-----------+----------+ |
* | /|\ . |
* | . . |
* | ChannelHandlerContext.fireIN_EVT() ChannelHandlerContext.OUT_EVT()|
* | [ method call] [method call] |
* | . . |
* | . \|/ |
* | +----------+----------+ +-----------+----------+ |
* | | Inbound Handler 2 | | Outbound Handler M-1 | |
* | +----------+----------+ +-----------+----------+ |
* | /|\ | |
* | | \|/ |
* | +----------+----------+ +-----------+----------+ |
* | | Inbound Handler 1 | | Outbound Handler M | |
* | +----------+----------+ +-----------+----------+ |
* | /|\ | |
* +---------------+-----------------------------------+---------------+
* | \|/
* +---------------+-----------------------------------+---------------+
* | | | |
* | [ Socket.read() ] [ Socket.write() ] |
* | |
* | Netty Internal I/O Threads (Transport Implementation) |
* +-------------------------------------------------------------------+
* </pre>2.2.4结论
1、不论解码器handler还是编码器handler，即接收的消息类型必须与待处理的消息类型一致，否则该handler不会被执行。
2、在解码器进行数据解码时，需要判断缓存区（ByteBuf）的数据是否足够，否则接收到的结果和期望结果可能不一致。
3.常用其他解码器
3.1解码器-ReplayingDecoder
（1）public abstract class ReplayingDecoder<S> extends ByteToMessageDecoder
（2）ReplayingDecoder扩展了ReplayingDecoder类，使用这个类，我们不必调用 readableBytes()方法。参数S指定了用户状态管理的类型，其中Void代表不需要状态管理。
（3）应用实例：使用 ReplayingDecoder 编写解码器，对前面的案例进行简化。
代码修改如下：
public class MyByteToLongDecoder2 extends ReplayingDecoder<Void> {
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
System.out.println("MyByteToLongDecoder2 的 decode()方法被调用");
// 在ReplayingDecoder 不需要判断数据是否足够读取，内部会进行处理判断
out.add(in.readLong());
}
}public class MyServerInitializer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
// 入站的 handler 进行解码
pipeline.addLast(new MyByteToLongDecoder2());
pipeline.addLast(new MyLongToByteEncoder());
pipeline.addLast(new MyServerHandler());
}
}public class MyClientInitializer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new MyLongToByteEncoder());
pipeline.addLast(new MyByteToLongDecoder2());
pipeline.addLast(new MyClientHandler());
}
}（4）ReplayingDecoder使用方便，但它也有一些局限性：
- 并不是所有的 ByteBuf 操作都被支持，如果调用了一个不被支持的方法，将会抛出一个UnsupportedOperationException。
- ReplayingDecoder 在某些情况下可能稍慢于 ByteToMessageDecoder，例如网络缓慢并且消息格式复杂时，消息会被拆成了多个碎片，速度变慢。
3.2解码器-LineBasedFrameDecoder
LineBasedFrameDecoder：这个类在 Netty 内部也有使用，它使用行尾控制字符（\n 或者 \r\n）作为分隔符来解析数据。
3.3解码器-DelimiterBasedFrameDecoder
DelimiterBasedFrameDecoder：使用自定义的特殊字符作为消息的分隔符。
3.4 解码器-HttpObjectDecoder
HttpObjectDecoder：一个 HTTP 数据的解码器。
3.5解码器-LengthFieldBasedFrameDecoder
LengthFieldBasedFrameDecoder：通过指定长度来标识整包消息，这样就可以自动的处理粘包和半包消息。
来源：https://blog.csdn.net/zhufei463738313/article/details/129438794