Netty源码分析-MessageToByteEncoder

Posted 2022-07-30 征服.刘华强

MessageToByteEncoder是一个抽象编码器，子类可重新encode方法把对象编码为ByteBuf输出。

 

源码分析

package io.netty.handler.codec;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelOutboundHandler;
import io.netty.channel.ChannelOutboundHandlerAdapter;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.ChannelPromise;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.internal.TypeParameterMatcher;


public abstract class MessageToByteEncoder<I> extends ChannelOutboundHandlerAdapter 

    private final TypeParameterMatcher matcher;
    private final boolean preferDirect;

   
    protected MessageToByteEncoder() 
        this(true);
    

    //匹配msg的数据类型是否为I，也就是子类定义的泛型
    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 
            //msg是I这种类型就调用编码逻辑，否则向下传递
            if (acceptOutboundMessage(msg)) 
                @SuppressWarnings("unchecked")
                I cast = (I) msg; //类型转换
                //分配ByteBuf，默认堆外类型
                buf = allocateBuffer(ctx, cast, preferDirect);
                try 
                    //子类实现编码逻辑
                    encode(ctx, cast, buf);
                 finally 
                    //释放msg
                    ReferenceCountUtil.release(cast);
                

                //如果buf可读，则写入ctx，实际就是交给netty调用底层socket输出
                if (buf.isReadable()) 
                    ctx.write(buf, promise);
                 else 
                    //否则释放buf
                    buf.release();
                    //向下传递空的buf对象
                    ctx.write(Unpooled.EMPTY_BUFFER, promise);
                
                buf = null;
             else 
                //如果不是I类型，则次编码器不做任何处理，直接传递msg对象给下一个编码器
                ctx.write(msg, promise);
            
         catch (EncoderException e) 
            throw e;
         catch (Throwable e) 
            throw new EncoderException(e);
         finally 
            if (buf != null) 
                buf.release();
            
        
    

   
    protected ByteBuf allocateBuffer(ChannelHandlerContext ctx, @SuppressWarnings("unused") I msg,
                               boolean preferDirect) throws Exception 
        if (preferDirect) 
            return ctx.alloc().ioBuffer();
         else 
            return ctx.alloc().heapBuffer();
        
    

    
    //需要子类实现，根据msg的内容，把数据写入ByteBuf中
    protected abstract void encode(ChannelHandlerContext ctx, I msg, ByteBuf out) throws Exception;

    //默认使用堆外内存
    protected boolean isPreferDirect() 
        return preferDirect;
    

 

例子，一个Integer类型的编码，负责把一个Integer写入到ByteBuf当中。

public class IntegerEncoder extends MessageToByteEncoder<Integer> 
          @Override
         public void encode(ChannelHandlerContext ctx, Integer msg, ByteBuf out)
                 throws Exception 
             out.writeInt(msg);