Netty 编解码器和 handler 的调用机制

1.基本说明
1) netty 的组件设计: Netty 的主要组件有 ChannelEventLoopChannelFutureChannelHandlerChannelPipe
2) ChannelHandler 充当了处理入站和出站数据的应用程序逻辑的容器。 例如, 实现 ChannelInboundHandler 接口(或
   ChannelInboundHandlerAdapter) , 你就可以接收入站事件和数据, 这些数据会被业务逻辑处理。 当要给客户端
   发 送 响 应 时 , 也 可 以 从 ChannelInboundHandler 冲 刷 数 据 。 业 务 逻 辑 通 常 写 在 一 个 或 者 多 个
   ChannelInboundHandler 中。 ChannelOutboundHandler 原理一样, 只不过它是用来处理出站数据的
3) ChannelPipeline 提供了 ChannelHandler 链的容器。 以客户端应用程序为例, 如果事件的运动方向是从客户端到
  服务端的, 那么我们称这些事件为出站的, 即客户端发送给服务端的数据会通过 pipeline 中的一系列ChannelOutboundHandler, 并被这些 Handler 处理, 反之则称为入站的 

Netty 编解码器和 handler 的调用机制


  2.编码解码器 

1) Netty 发送或者接受一个消息的时候, 就将会发生一次数据转换。 入站消息会被解码: 从字节转换为另一种格式(比如 java 对象) ; 如果是出站消息, 它会被编码成字节。

2) Netty 提供一系列实用的编解码器, 他们都实现了 ChannelInboundHadnler 或者 ChannelOutboundHandler 接口。在这些类中, channelRead 方法已经被重写了。 以入站为例, 对于每个从入站 Channel 读取的消息, 这个方法会
被调用。 随后, 它将调用由解码器所提供的 decode()方法进行解码, 并将已经解码的字节转发给 ChannelPipeline中的下一个 ChannelInboundHandler。 

注意:进站和出站都是想对而言


 3 解码器-ByteToMessageDecoder

1) 关系继承图

Netty 编解码器和 handler 的调用机制

 2) 由于不可能知道远程节点是否会一次性发送一个完整的信息, tcp 有可能出现粘包拆包的问题, 这个类会对入站数据进行缓冲, 直到它准备好被处理.
 3) 一个关于 ByteToMessageDecoder 实例分析

Netty 编解码器和 handler 的调用机制

 Netty 编解码器和 handler 的调用机制


4 Netty handler 链的调用机制

实例要求:
1) 使用自定义的编码器和解码器来说明 Netty handler 调用机制
  客户端发送 long -> 服务器
  服务端发送 long -> 客户端
2) 案例演示

Netty 编解码器和 handler 的调用机制

 3) 结论

  不论解码器 handler 还是 编码器 handler 即接收的消息类型必须与待处理的消息类型一致, 否则该 handler 不会被执行
  在解码器 进行数据解码时, 需要判断 缓存区(ByteBuf)的数据是否足够 , 否则接收到的结果会期望结果可能不一致 

代码:

MyServer
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
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).channel(NioServerSocketChannel.class).childHandler(new MyServerInitializer()); //自定义一个初始化类


            ChannelFuture channelFuture = serverBootstrap.bind(7000).sync();
            channelFuture.channel().closeFuture().sync();

        }finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }

    }
}
MyServerInitializer
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;


public class MyServerInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();//一会下断点

        //入站的handler进行解码 MyByteToLongDecoder
        //pipeline.addLast(new MyByteToLongDecoder());
        pipeline.addLast(new MyByteToLongDecoder2());
        //出站的handler进行编码
        pipeline.addLast(new MyLongToByteEncoder());
        //自定义的handler 处理业务逻辑
        pipeline.addLast(new MyServerHandler());
        System.out.println("xx");
    }
}
MyServerHandler
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

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(98765L);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}
MyClient
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

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
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;


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 MyByteToLongDecoder());
        pipeline.addLast(new MyByteToLongDecoder2());
        //加入一个自定义的handler , 处理业务
        pipeline.addLast(new MyClientHandler());


    }
}
MyClientHandler
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.CharsetUtil;

import java.nio.charset.Charset;

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);

    }

    //重写channelActive 发送数据

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("MyClientHandler 发送数据");
        //ctx.writeAndFlush(Unpooled.copiedBuffer(""))
        ctx.writeAndFlush(123456L); //发送的是一个long

        //分析
        //1. "abcdabcdabcdabcd" 是 16个字节
        //2. 该处理器的前一个handler 是  MyLongToByteEncoder
        //3. MyLongToByteEncoder 父类  MessageToByteEncoder
        //4. 父类  MessageToByteEncoder
        /*

         public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        ByteBuf buf = null;
        try {
            if (acceptOutboundMessage(msg)) { //判断当前msg 是不是应该处理的类型,如果是就处理,不是就跳过encode
                @SuppressWarnings("unchecked")
                I cast = (I) msg;
                buf = allocateBuffer(ctx, cast, preferDirect);
                try {
                    encode(ctx, cast, buf);
                } finally {
                    ReferenceCountUtil.release(cast);
                }

                if (buf.isReadable()) {
                    ctx.write(buf, promise);
                } else {
                    buf.release();
                    ctx.write(Unpooled.EMPTY_BUFFER, promise);
                }
                buf = null;
            } else {
                ctx.write(msg, promise);
            }
        }
        4. 因此我们编写 Encoder 是要注意传入的数据类型和处理的数据类型一致
        */
       // ctx.writeAndFlush(Unpooled.copiedBuffer("abcdabcdabcdabcd",CharsetUtil.UTF_8));

    }
}
MyByteToLongDecoder
解码器
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;

import java.util.List;

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 被调用");
        //因为 long 8个字节, 需要判断有8个字节,才能读取一个long
        if(in.readableBytes() >= 8) {
            out.add(in.readLong());
        }
    }
}

解码器-ReplayingDecoder

 1)  public abstract class ReplayingDecoder<S> extends ByteToMessageDecoder
 2)  ReplayingDecoder 扩展了 ByteToMessageDecoder 类, 使用这个类, 我们不必调用 readableBytes()方法。 参数 S指定了用户状态管理的类型, 其中 Void 代表不需要状态管理
 3)  应用实例: 使用 ReplayingDecoder 编写解码器, 对前面的案例进行简化 [案例演示 

MyByteToLongDecoder2
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ReplayingDecoder;

import java.util.List;

public class MyByteToLongDecoder2 extends ReplayingDecoder<Void> {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {

        System.out.println("MyByteToLongDecoder2 被调用");
        //在 ReplayingDecoder 不需要判断数据是否足够读取,内部会进行处理判断
        out.add(in.readLong());


    }
}

4)  ReplayingDecoder 使用方便, 但它也有一些局限性:
  1. 并 不 是 所 有 的 ByteBuf 操 作 都 被 支 持 , 如 果 调 用 了 一 个 不 被 支 持 的 方 法 , 将 会 抛 出 一 个UnsupportedOperationException
  2. ReplayingDecoder 在某些情况下可能稍慢于 ByteToMessageDecoder, 例如网络缓慢并且消息格式复杂时,
      消息会被拆成了多个碎片, 速度变慢 。


6 其它编解码器

1) LineBasedFrameDecoder: 这个类在 Netty 内部也有使用, 它使用行尾控制字符(\n 或者\r\n) 作为分隔符来解析数据。
2) DelimiterBasedFrameDecoder: 使用自定义的特殊字符作为消息的分隔符。
3) HttpObjectDecoder: 一个 HTTP 数据的解码器
4) LengthFieldBasedFrameDecoder: 通过指定长度来标识整包消息, 这样就可以自动的处理黏包和半包消息。


7 其它编码器

Netty 编解码器和 handler 的调用机制