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

基本说明

netty的组件设计：Netty的主要组件有Channel、EventLoop、ChannelFuture、ChannelHandler、ChannelPipe等
ChannelHandler充当了处理入站和出站数据的应用程序逻辑的容器。如：实现ChannelInboundHandler接口(或ChannelInboundHandlerAdapter)，就可以接收入站事件和数据，这些数据会被业务逻辑处理。当要给客户端发送响应时，也可以从ChannelInboundHandler冲刷数据。业务逻辑通常写在一个或多个ChannelInboundHandler中。ChannelOutboundHandler原理一样，只不过它是用来处理出站数据的
ChannelPipeline提供了ChannelHandler链的容器。以客户端应用程序为例，如果事件的运动方向是从客户端到服务端的，那么称这些事件为出站的，即客户端发送给服务端的数据会通过pipeline中的一系列ChannelOutboundHandler，并被这些Handler处理，反之则称为入站的(从客户端角度判断是出/入站)

出站要编码，入站要解码

出站对应写，入站对应读

编码解码器

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

解码器-ByteToMessageDecoder

关系继承图
由于不可能知道远程节点是否会一次性发送一个完整的信息，tcp有可能出现粘包拆包的问题，这个类会对入站数据进行缓冲，直到它准备好被处理

一个关于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());
        }
    }
}

说明：

这个例子，每次入站从ByteBuf中读取4字节，将其解码为一个int，然后将它添加到下一个List中，当没有更多元素可以被添加到该List中时，它的内容将会被发送给下一个ChannelInboundHandler。int在被添加到List中时，会被自动装箱为Integer。在调用readInt()方法前必须验证所输入的ByteBuf是否具有足够的数据

decode执行分析图

解码器-ReplayingDecoder

public abstract class ReplayingDecoder< S > extends ByteToMessageDecoder
ReplayingDecoder扩展了ByteToMessageDecoder类，使用这个类，我们不必调用readableBytes()方法。参数S指定了用户状态管理的类型，其中Void代表不需要状态管理

应用实例

class MyByteToLongDecoder2 extends ReplayingDecoder<Void>{

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        // 在ReplayingDecoder不需要判断数据是否足够读取，内部会进行处理判断
        out.add(in.readLong());
    }
}

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

其它解码器

LineBasedFrameDecoder：这个类在Netty内部也有使用，它使用行尾控制字符()作为分隔符来解析数据
DelimiterBasedFrameDecoder：使用自定义的特殊字符作为消息的分隔符
HttpObjectDecoder：一个HTTP数据的解码器
LengthFieldBasedFrameDecoder：通过指定长度来标识整包消息，这样就可以自动处理粘包和半包消息

Netty的handler调用机制

实例

示意图

代码

public class MyServer {
    public static void main(String[] args) throws InterruptedException {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup,workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new MyServerInitializer());
            ChannelFuture channelFuture = b.bind(7000).sync();
            channelFuture.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}
class MyServerInitializer extends ChannelInitializer<SocketChannel>{

    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        // 入站handler解码
        pipeline.addLast(new MyByteToLongDecoder());
        // 出站handler编码
        pipeline.addLast(new MyLongToByteEncoder());

        pipeline.addLast(new MyServerHandler());
    }
}
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 {
        if(in.readableBytes() >= 8){
            out.add(in.readLong());
        }
    }
}

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

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

public class MyClient {
    public static void main(String[] args) throws InterruptedException {
        EventLoopGroup group = new NioEventLoopGroup();

        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new MyClientInitializer());
            ChannelFuture channelFuture = b.connect("127.0.0.1", 7000).sync();
            channelFuture.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
    }
}
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());
    }
}
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);
    }
}
class MyClientHandler extends SimpleChannelInboundHandler<Long>{

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Long msg) throws Exception {
        System.out.println("服务器"+ctx.channel().remoteAddress() + " 回送消息："+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. 父类中的write方法中会判断数据是否应该处理的类型，如果不是则会跳过encode
         * 5. 因此编写Encoder需要注意传入的数据类型和处理的数据类型一致
         */
        ctx.writeAndFlush(Unpooled.copiedBuffer("abcdabcdabcdabcd", CharsetUtil.UTF_8));

    }
}

结论

不论解码器handler还是编码器handler即接收的消息类型必须与待处理的消息类型一致，否则该handler不会被执行

在解码器进行数据解码时，需要判断缓存区(ByteBuf)的数据是否足够，否则接收到的结果可能会和期望结果不一致

Log4j整合到Netty

在Maven中添加对Log4j的依赖
配置Log4j，在resources/log4j.properties