Netty源码分析-ProtobufEncoder-Decoder

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

ProtobufEncoder源码分析

@Sharable
public class ProtobufEncoder extends MessageToMessageEncoder<MessageLiteOrBuilder> 
    @Override
    protected void encode(ChannelHandlerContext ctx, MessageLiteOrBuilder msg, List<Object> out)
            throws Exception 
        //当前对象如果是google protocol buffer生成的对象，他们都实现了该接口
        if (msg instanceof MessageLite) 
            //把byte[]包装了Bytebuf返回
            out.add(wrappedBuffer(((MessageLite) msg).toByteArray()));
            return;
        
        if (msg instanceof MessageLite.Builder) 
            out.add(wrappedBuffer(((MessageLite.Builder) msg).build().toByteArray()));
        
    

 

ProtobufDecoder源码分析

package io.netty.handler.codec.protobuf;

import com.google.protobuf.ExtensionRegistry;
import com.google.protobuf.ExtensionRegistryLite;
import com.google.protobuf.Message;
import com.google.protobuf.MessageLite;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.MessageToMessageDecoder;

import java.util.List;


@Sharable
public class ProtobufDecoder extends MessageToMessageDecoder<ByteBuf> 

    private static final boolean HAS_PARSER;

    static 
        boolean hasParser = false;
        try 
            // MessageLite.getParserForType() is not available until protobuf 2.5.0.
            MessageLite.class.getDeclaredMethod("getParserForType");
            hasParser = true;
         catch (Throwable t) 
            // Ignore
        

        HAS_PARSER = hasParser;
    

    private final MessageLite prototype;
    private final ExtensionRegistryLite extensionRegistry;

    /**
     * Creates a new instance.
     */
    public ProtobufDecoder(MessageLite prototype) 
        this(prototype, null);
    

    public ProtobufDecoder(MessageLite prototype, ExtensionRegistry extensionRegistry) 
        this(prototype, (ExtensionRegistryLite) extensionRegistry);
    

    public ProtobufDecoder(MessageLite prototype, ExtensionRegistryLite extensionRegistry) 
        if (prototype == null) 
            throw new NullPointerException("prototype");
        
        this.prototype = prototype.getDefaultInstanceForType();
        this.extensionRegistry = extensionRegistry;
    

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out)
            throws Exception 
        final byte[] array;
        final int offset;
        final int length = msg.readableBytes();
        //判断msg内部是否包含了byte数组
        if (msg.hasArray()) 
        	//直接去出byte数组
            array = msg.array();
            //offset是数组的起始位置，readerIndex是数组有效字节
            offset = msg.arrayOffset() + msg.readerIndex();
         else 
        	//创建一个length长度的数组
            array = new byte[length];
            //把底层字节写入数组
            msg.getBytes(msg.readerIndex(), array, 0, length);
            offset = 0;
        

        //根据protocol的API版本选择反序列化代码执行
        //将字节数组反序列化为protocol的实例
        if (extensionRegistry == null) 
            if (HAS_PARSER) 
                out.add(prototype.getParserForType().parseFrom(array, offset, length));
             else 
                out.add(prototype.newBuilderForType().mergeFrom(array, offset, length).build());
            
         else 
            if (HAS_PARSER) 
                out.add(prototype.getParserForType().parseFrom(
                        array, offset, length, extensionRegistry));
             else 
                out.add(prototype.newBuilderForType().mergeFrom(
                        array, offset, length, extensionRegistry).build());
            
        
    

 

编解码器应用，需要配合LengthFieldBasedFrameDecoder和LengthFieldPrepender，它俩使用定长4个字节写入消息长度。

 // Decoders
 pipeline.addLast("frameDecoder",
                  new LengthFieldBasedFrameDecoder(1048576, 0, 4, 0, 4));
 pipeline.addLast("protobufDecoder",
                  new ProtobufDecoder(MyMessage.getDefaultInstance()));

 // Encoder
 pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
 pipeline.addLast("protobufEncoder", new ProtobufEncoder());

还可以使用ProtobufVarint32FrameDecoder和ProtobufVarint32LengthFieldPrepender，它俩采用google的varint32编码包头长度字节。

ch.pipeline().addLast(new ProtobufVarint32FrameDecoder());
ch.pipeline().addLast(new ProtobufDecoder(RichManProto.RichMan.getDefaultInstance()));


ch.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender());
ch.pipeline().addLast(new ProtobufEncoder());