netty源码之ChannelHandler

超、凢脫俗 2022-12-28 07:22 123阅读 0赞

# netty源码之ChannelHandler #

ChannelHandler类似于Servlet的Filter过滤器，负责对I/O事件或者IO操作进行拦截和处理，它可以选择性地拦截和处理自己感兴趣的事件，也可以透传和终止事件的传递。

基于ChannelHandler接口，用户可以方便地进行业务逻辑定制，例如打印日志、统一封装异常信息、性能统计和消息编解码等。

## ChannelHandler ##

ChannelHandler的继承关系：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center]

由于ChannelHandler是Netty框架和用户代码的主要扩展和定制点，所以它的子类种类繁多、功能各异，系统ChannelHandler主要分类如下:

*  ChannelPipeline中的系统ChannelHandler，用于I/O操作和对事件进行预处理，对于用户不可见，这类ChannelHandler主要包括HeadHandle和TailHandler。
 *  编解码ChannelHandler，包括ByteToMessageCodec、MessageToMessageDecoder等，这些编解码类本身又包含多种子类。
 *  其他系统功能性ChannelHandler，包括读写超时Handler、日志Handler等。
 *  协议编解码Handler，Netty事先实现了很多协议的ChannelHandler，比如Http协议相关、SSL相关等等。

## ChannelHandlerAdapter、ChannelInboundHandlerAdapter、ChannelOutboundHandlerAdapter ##

对于大多数的ChannelHandler会选择性地拦截和处理某个或者某些事件，其他的事件会忽略，由下一个ChannelHandler进行拦截和处理。这就会导致一个问题：用户ChannelHandler必须要实现ChannelHandler的所有接口，包括它不关心的那些事件处理接口，这会导致用户代码的冗余和臃肿，代码的可维护性也会变差。

为了解决这个问题，Netty提供了ChannelHandlerAdapter基类，它的实现要么就是事件透传，要么就是空实现，如果用户ChannelHandler关心某个事件，只需要覆盖ChannelHandlerAdapter对应的方法即可，对于不关心的，可以直接继承使用父类的方法，这样子类的代码就会非常简洁和清晰。

我们的ChannelHandler都是直接继承自ChannelHandlerAdapter、ChannelInboundHandlerAdapter、ChannelOutboundHandlerAdapter，开发起来非常简单和高效。

## channelRead与channelReadComplete的区别 ##

下面用代码来验证一下：

Server端添加如下Handler：

ch.pipeline().addLast(new LoggingHandler());
    ch.pipeline().addLast(new LineBasedFrameDecoder(128));
    ch.pipeline().addLast(new ServerHandler());

ServerHandler中的代码如下：

@Slf4j
    public class ServerHandler extends ChannelInboundHandlerAdapter { 
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
            ByteBuf byteBuf = (ByteBuf) msg;
            log.info("channelRead: {}", byteBuf.toString(CharsetUtil.UTF_8));
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { 
            log.info("channelReadComplete");
        }
    }

ClientHandler端的代码如下：

public class ClientHandler extends ChannelInboundHandlerAdapter { 
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception { 
    
            for (int i = 0; i < 12; i++) { 
                ByteBuf sendByteBuf = Unpooled.copiedBuffer(("helloworldnetty\n").getBytes());
                ctx.writeAndFlush(sendByteBuf);
            }
        }
    }

Server收到的数据如下：

2020-12-01 17:05:21,119 DEBUG [nioEventLoopGroup-3-1] (AbstractInternalLogger.java:145) - [id: 0x7b4bdd32, L:/127.0.0.1:8899 - R:/127.0.0.1:12903] READ: 192B
             +-------------------------------------------------+
             |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
    +--------+-------------------------------------------------+----------------+
    |00000000| 68 65 6c 6c 6f 77 6f 72 6c 64 6e 65 74 74 79 0a |helloworldnetty.|
    |00000010| 68 65 6c 6c 6f 77 6f 72 6c 64 6e 65 74 74 79 0a |helloworldnetty.|
    |00000020| 68 65 6c 6c 6f 77 6f 72 6c 64 6e 65 74 74 79 0a |helloworldnetty.|
    |00000030| 68 65 6c 6c 6f 77 6f 72 6c 64 6e 65 74 74 79 0a |helloworldnetty.|
    |00000040| 68 65 6c 6c 6f 77 6f 72 6c 64 6e 65 74 74 79 0a |helloworldnetty.|
    |00000050| 68 65 6c 6c 6f 77 6f 72 6c 64 6e 65 74 74 79 0a |helloworldnetty.|
    |00000060| 68 65 6c 6c 6f 77 6f 72 6c 64 6e 65 74 74 79 0a |helloworldnetty.|
    |00000070| 68 65 6c 6c 6f 77 6f 72 6c 64 6e 65 74 74 79 0a |helloworldnetty.|
    |00000080| 68 65 6c 6c 6f 77 6f 72 6c 64 6e 65 74 74 79 0a |helloworldnetty.|
    |00000090| 68 65 6c 6c 6f 77 6f 72 6c 64 6e 65 74 74 79 0a |helloworldnetty.|
    |000000a0| 68 65 6c 6c 6f 77 6f 72 6c 64 6e 65 74 74 79 0a |helloworldnetty.|
    |000000b0| 68 65 6c 6c 6f 77 6f 72 6c 64 6e 65 74 74 79 0a |helloworldnetty.|
    +--------+-------------------------------------------------+----------------+
    2020-12-01 17:05:21,128  INFO [nioEventLoopGroup-3-1] (ServerHandler.java:15) - channelRead: helloworldnetty
    2020-12-01 17:05:21,129  INFO [nioEventLoopGroup-3-1] (ServerHandler.java:15) - channelRead: helloworldnetty
    2020-12-01 17:05:21,129  INFO [nioEventLoopGroup-3-1] (ServerHandler.java:15) - channelRead: helloworldnetty
    2020-12-01 17:05:21,129  INFO [nioEventLoopGroup-3-1] (ServerHandler.java:15) - channelRead: helloworldnetty
    2020-12-01 17:05:21,129  INFO [nioEventLoopGroup-3-1] (ServerHandler.java:15) - channelRead: helloworldnetty
    2020-12-01 17:05:21,129  INFO [nioEventLoopGroup-3-1] (ServerHandler.java:15) - channelRead: helloworldnetty
    2020-12-01 17:05:21,129  INFO [nioEventLoopGroup-3-1] (ServerHandler.java:15) - channelRead: helloworldnetty
    2020-12-01 17:05:21,130  INFO [nioEventLoopGroup-3-1] (ServerHandler.java:15) - channelRead: helloworldnetty
    2020-12-01 17:05:21,130  INFO [nioEventLoopGroup-3-1] (ServerHandler.java:15) - channelRead: helloworldnetty
    2020-12-01 17:05:21,130  INFO [nioEventLoopGroup-3-1] (ServerHandler.java:15) - channelRead: helloworldnetty
    2020-12-01 17:05:21,130  INFO [nioEventLoopGroup-3-1] (ServerHandler.java:15) - channelRead: helloworldnetty
    2020-12-01 17:05:21,130  INFO [nioEventLoopGroup-3-1] (ServerHandler.java:15) - channelRead: helloworldnetty
    2020-12-01 17:05:21,130 DEBUG [nioEventLoopGroup-3-1] (AbstractInternalLogger.java:145) - [id: 0x7b4bdd32, L:/127.0.0.1:8899 - R:/127.0.0.1:12903] READ COMPLETE
    2020-12-01 17:05:21,130  INFO [nioEventLoopGroup-3-1] (ServerHandler.java:20) - channelReadComplete

从运行结果可以发现channelRead执行了12次，而channelReadComplete只执行了一次。

总结两者的区别：

*  channelReadComplete：每次读取完Socket的接收缓冲区的报文，channelReadComplete会被触发一次，注意这里是Socket的接收缓冲区，不是Netty中接收的ByteBuf的大小。
 *  channelRead：从Netty的设计角度和实现来看，当收到一个完整的应用层消息报文，channelRead会被触发一次，也就是读取到一条你需要的完整数据就会触发一次。

如果每次传递的完整的数据很小，那么ByteBuf中就会存储多个完整的数据，那么channelRead就会被回调多次，如果传递的完整的数据很大，那么有可能存在调用三次channelReadComplete才会调用一次channelRead的情况。

## ByteToMessageDecoder、MessageToByteEncoder ##

ByteToMessageDecoder和MessageToByteEncoder是我们用的比较多的两个ChannelHandler的子类，我们挑这两个子类来看看里面关键的方法。

### ByteToMessageDecoder ###

ByteToMessageDecoder用于将ByteBuf转换成Message，message可以是POJO等等，转换后继续在ChannelPipeline中往后面的入站Handler传递，ByteToMessageDecoder继承自ChannelInboundHandlerAdapter，需要开发者实现的是decode方法：

protected abstract void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception;

in是用来读取的数据，out是我们转换后的对象列表。

ByteToMessage是ChannelInboundHandler，很明显，关键的方法是channelRead：

// io.netty.handler.codec.ByteToMessageDecoder#channelRead
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
        if (msg instanceof ByteBuf) { 
            CodecOutputList out = CodecOutputList.newInstance();
            try { 
                first = cumulation == null;
                cumulation = cumulator.cumulate(ctx.alloc(),
                        first ? Unpooled.EMPTY_BUFFER : cumulation, (ByteBuf) msg);
                callDecode(ctx, cumulation, out);
            } catch (DecoderException e) { 
                throw e;
            } catch (Exception e) { 
                throw new DecoderException(e);
            } finally { 
                try { 
                    if (cumulation != null && !cumulation.isReadable()) { 
                        numReads = 0;
                        cumulation.release();
                        cumulation = null;
                    } else if (++numReads >= discardAfterReads) { 
                        // We did enough reads already try to discard some bytes so we not risk to see a OOME.
                        // See https://github.com/netty/netty/issues/4275
                        numReads = 0;
                        discardSomeReadBytes();
                    }
    
                    int size = out.size();
                    firedChannelRead |= out.insertSinceRecycled();
                    fireChannelRead(ctx, out, size);
                } finally { 
                    out.recycle();
                }
            }
        } else { 
            ctx.fireChannelRead(msg);
        }
    }

从总体结构来说，

if (msg instanceof ByteBuf) { 
    // （略）这里是第二层结构
    } else { 
    	ctx.fireChannelRead(msg);
    }

很明显，判断传入类型是否为ByteBuf，类型符合则处理，不符合则透传。

在具体的处理部分中，创建一个 CodecOutputList(也是一种 list)保存转码后的对象列表。因为有粘包和半包需要处理，使用ByteBuf型的cumulation来保留处理不完的ByteBuf，同时使用一个专门的接口Cumulator来负责合并。

之后调用callDecode(ctx, cumulation, out)，callDecode中循环调用我们要实现的抽象方法decode(ctx,in,out)来解码。然后将解码后的对象通过fireChannelRead交给后面的Handler进行处理。

### MessageToByteEncoder ###

MessageToByteEncoder用于将Message转换成ByteBuf，转换后继续在ChannelPipeline中往前面的出站Handler传递，MessageToByteEncoder继承自ChannelOutboundHandlerAdapter，需要开发者实现的是encode方法。

protected abstract void encode(ChannelHandlerContext ctx, I msg, ByteBuf out) throws Exception;

MessageToByteEncoder是一个ChannelOutboundHandler，里面关键的就是write方法：

public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { 
        ByteBuf buf = null;
        try { 
            if (acceptOutboundMessage(msg)) { 
                @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);
            }
        } catch (EncoderException e) { 
            throw e;
        } catch (Throwable e) { 
            throw new EncoderException(e);
        } finally { 
            if (buf != null) { 
                buf.release();
            }
        }
    }

其中的逻辑是比较简单的，判断消息是否是类型匹配的，不是就直接往前面的出站handler传递，是的话会申请一个缓冲区，然后调用开发者实现encode方法进行编码，编码成功，通过ctx往前面的出站Handler进行传递，并会释放消息。如果编码失败，就往前写一个空缓冲区，把申请的缓冲区释放了。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/2181745bb76741f1b2feb4dc9584534e.png