Netty中的ChannelHandler的基础知识

Posted 2023-02-02 javartisan

Netty ChannelHandler概览     

Netty中的ChannelHandler类似于工程对产品流水线生产中：生产线上的每一步的处理器，而生产线则就是Netty中的ChannelPipeline。Netty也正是通过ChannelHandler实现了业务与底层网络的解耦。Netty中的ChannelHandler按照输出字节流向分为In,Out,Duplex三种ChannelHandler，本文主要介绍前面两种，Netty中ChannelHandler关系图如下：

• ChannelHander是顶级抽象，基础接口类。
• ChannelHandlerAdapter则是ChannelHandler的一个适配器，对ChannelHandler添加了新的行为，用于判断当前Handler是否执行共享，ChannelHandler共享是指：如果一个ChannelHandler支持共享，则该ChannelHandler可以被添加到多个ChannelPipeline中。
• ChannelInboundHandler、ChannelOutboundHandler则是分别对应进站与出站的ChannelHandler处理器。
• ChannelInboundHandlerAdpater、ChannelOutboundHandlerAdapter分别实现了对应类型的ChannelHandler之外，还实现了ChannelHandlerAdpater接口，这样就有了判断是否是共享ChannelHandler的行为。

ChannelInboundHandler

Netty中无论对应客户端还是服务器端都有读、写操作，读操作对应的就是进站操作，进站的数据都会经过ChannelInboundHandler处理，例如解码器都是ChannelInboundHandler的子类实现；本文简单介绍一下SimpleChannelInboundHandler。

SimpleChannelInboundHandler是ChannelInboundHandlerAdpater子类实现，SimpleChannelInboundHandler是一个类型敏感的处理器，在明确上游消息类型时候可以使用该处理器，这样就避免了类型的强转；在SimpleChannelInboundHandler中新增加两个方法分别是：acceptInboundMessage与channelRead0方法，同时重写了channelRead方法，在channelRead方法中进行处理类型转换，对于下游事件处理通过重写channelRead0即可，具体源码如下：

    /**
     * Returns @code true if the given message should be handled. If @code false it will be passed to the next
     * @link ChannelInboundHandler in the @link ChannelPipeline.
     */
    public boolean acceptInboundMessage(Object msg) throws Exception 
        return matcher.match(msg);
    

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception 
        boolean release = true;
        try 
            if (acceptInboundMessage(msg)) 
                @SuppressWarnings("unchecked")
                I imsg = (I) msg;
                channelRead0(ctx, imsg);
             else 
                release = false;
                ctx.fireChannelRead(msg);
            
         finally 
            if (autoRelease && release) 
                ReferenceCountUtil.release(msg);
            
        
    

    /**
     * <strong>Please keep in mind that this method will be renamed to
     * @code messageReceived(ChannelHandlerContext, I) in 5.0.</strong>
     *
     * Is called for each message of type @link I.
     *
     * @param ctx           the @link ChannelHandlerContext which this @link SimpleChannelInboundHandler
     *                      belongs to
     * @param msg           the message to handle
     * @throws Exception    is thrown if an error occurred
     */
    protected abstract void channelRead0(ChannelHandlerContext ctx, I msg) throws Exception;

备注：ChannelInitializer也是一个入站处理器，当Channel建立时候进行一些初始化工作，例如Pipeline上面添加Handler操作。

ChannelOutboundHandler

Netty中Channel上的写操作对应着出站操作，通常流经ChannelOutboundHandler处理之后通过网络将数据发出去；ChannelOutboundHandler的子类实现一般都是编码器，对应编码操作。

MessageToByteEncoder是一个类型敏感的编码器，将指定类型的Message序列化为二进制数据；该类新增三个方法分别是：acceptOutboundMessage（用于判断类型），allocateBuffer（分配缓冲区），encode（编码逻辑供子类具体实现）；而write方法中进行类型的处理，判断是否类型匹配，匹配则进行encode处理。

    /**
     * Returns @code true if the given message should be handled. If @code false it will be passed to the next
     * @link ChannelOutboundHandler in the @link ChannelPipeline.
     */
    public boolean acceptOutboundMessage(Object msg) throws Exception 
        return matcher.match(msg);
    

    @Override
    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();
            
        
    

    /**
     * Allocate a @link ByteBuf which will be used as argument of @link #encode(ChannelHandlerContext, I, ByteBuf).
     * Sub-classes may override this method to return @link ByteBuf with a perfect matching @code initialCapacity.
     */
    protected ByteBuf allocateBuffer(ChannelHandlerContext ctx, @SuppressWarnings("unused") I msg,
                               boolean preferDirect) throws Exception 
        if (preferDirect) 
            //Direct Buffer
            return ctx.alloc().ioBuffer();
         else 
            //Heap Buffer
            return ctx.alloc().heapBuffer();
        
    

    /**
     * Encode a message into a @link ByteBuf. This method will be called for each written message that can be handled
     * by this encoder.
     *
     * @param ctx           the @link ChannelHandlerContext which this @link MessageToByteEncoder belongs to
     * @param msg           the message to encode
     * @param out           the @link ByteBuf into which the encoded message will be written
     * @throws Exception    is thrown if an error occurs
     */
    protected abstract void encode(ChannelHandlerContext ctx, I msg, ByteBuf out) throws Exception;

ChannelDuplexHandler

经过如上的进站、出站ChannelHandler的理解基础之上，对于ChannelDuplexHandler理解就比较容易了，ChannelDuplexHandler是一个双工的处理器，既可以处理入站信息，也可以处理出站信息，通常对应的也就是编解码器（Codec-编解码器）。

Netty中的MessageToMessageCodec就是一个典型的编解码器，该编解码器新增加了4个方法，分别进站对应两个，出站对应两个，对于encode与decode由子类具体实现响应的编码与解码。如下是一个简单的Long与Integer类型之间编码解码的实现：

public class NumberCodec extends MessageToMessageCodec<Integer, Long> 
    @Override
    public Long decode(ChannelHandlerContext ctx, Integer msg, List<Object> out)
            throws Exception 
        out.add(msg.longValue());
    

    @Override
    public Integer encode(ChannelHandlerContext ctx, Long msg, List<Object> out)
            throws Exception 
        out.add(msg.intValue());