以dubbo使用netty4为通信框架来进行分析。

客户端请求编码总体流程如下：

1 NettyCodecAdapter$InternalEncoder.encode(ChannelHandlerContext ctx, Channel ch, Object msg) 2 -->new NettyBackedChannelBuffer(ByteBuf buffer) // 创建一个buffer 3 -->NettyChannel.getOrAddChannel(io.netty.channel.Channel ch, URL url, ChannelHandler handler) 4 -->DubboCountCodec.encode(Channel channel, ChannelBuffer buffer, Object msg) 5   -->ExchangeCodec.encode(Channel channel, ChannelBuffer buffer, Object msg) 6       -->encodeRequest(Channel channel, ChannelBuffer buffer, Request req) 7         -->getSerialization(Channel channel)   //获取Hessian2Serialization序列化实例 8           -->CodecSupport.getSerialization(URL url) 9             -->ExtensionLoader.getExtensionLoader(Serialization.class).getExtension(url.getParameter("serialization", "hessian2"))10         
     11         -->byte[] header = new byte[16]12         -->Bytes.short2bytes(MAGIC, header)  //设置前两个字节为魔数[-38, -69, 0, ..., 0]13         
     14         -->header[2] = (byte) (FLAG_REQUEST | serialization.getContentTypeId());15          if (req.isTwoWay()) header[2] |= FLAG_TWOWAY;16          if (req.isEvent()) header[2] |= FLAG_EVENT;17       
     18       -->Bytes.long2bytes(req.getId(), header, 4);19       
     20       -->new Hessian2ObjectOutput(out)21       -->DubboCodec.encodeRequestData(Channel channel, ObjectOutput out, Object data)22       -->Bytes.int2bytes(len, header, 12); // 设置第13~16个字节（int是32位，4个字节）：消息体长度23       -->buffer.writeBytes(header); // 将header写入buffer的前16位

总体流程很简单：

• 创建一个buffer
• 创建一个16位的byte[16] header，将魔数、请求标志、序列化协议ID、twoway/event标志、requestID、请求体长度写入header
• 之后序列化请求体，从buffer的第17位向后写入序列化后的请求体字节数组
• 最后，将header中的内容写入buffer的前16位
• 最后发送buffer

 首先来看一下netty编解码的入口：com.alibaba.dubbo.remoting.transport.netty4：

1     @Override 2     protected void doOpen() throws Throwable { 3         NettyHelper.setNettyLoggerFactory(); 4         final NettyClientHandler nettyClientHandler = new NettyClientHandler(getUrl(), this); 5         bootstrap = new Bootstrap(); 6         bootstrap.group(nioEventLoopGroup) 7                 .option(ChannelOption.SO_KEEPALIVE, true) 8                 .option(ChannelOption.TCP_NODELAY, true) 9                 .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)10                 //.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, getTimeout())11                 .channel(NioSocketChannel.class);12 13         if (getTimeout() < 3000) {14             bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000);15         } else {16             bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, getTimeout());17         }18 19         bootstrap.handler(new ChannelInitializer() {20 21             protected void initChannel(Channel ch) throws Exception {22                 NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyClient.this);23                 ch.pipeline()//.addLast("logging",new LoggingHandler(LogLevel.INFO))//for debug24                         .addLast("decoder", adapter.getDecoder())25                         .addLast("encoder", adapter.getEncoder())26                         .addLast("handler", nettyClientHandler);27             }28         });29     }

NettyCodecAdapter：

1 final class NettyCodecAdapter { 2     private final ChannelHandler encoder = new InternalEncoder(); 3     private final ChannelHandler decoder = new InternalDecoder(); 4     private final Codec2 codec; 5     private final URL url; 6     private final com.alibaba.dubbo.remoting.ChannelHandler handler; 7  8     public NettyCodecAdapter(Codec2 codec, URL url, com.alibaba.dubbo.remoting.ChannelHandler handler) { 9         this.codec = codec;10         this.url = url;11         this.handler = handler;12     }13 14     public ChannelHandler getEncoder() {15         return encoder;16     }17 18     public ChannelHandler getDecoder() {19         return decoder;20     }21 22     private class InternalEncoder extends MessageToByteEncoder {23         protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception {24             com.alibaba.dubbo.remoting.buffer.ChannelBuffer buffer = new NettyBackedChannelBuffer(out);25             Channel ch = ctx.channel();26             NettyChannel channel = NettyChannel.getOrAddChannel(ch, url, handler);27             try {28                 codec.encode(channel, buffer, msg);29             } finally {30                 NettyChannel.removeChannelIfDisconnected(ch);31             }32         }33     }34 35     private class InternalDecoder extends ByteToMessageDecoder {36         protected void decode(ChannelHandlerContext ctx, ByteBuf input, List

一、创建ChannelBuffer

1 com.alibaba.dubbo.remoting.buffer.ChannelBuffer buffer = new NettyBackedChannelBuffer(out);

这里的out是：

1 ByteBuf buffer = SimpleLeakAwareByteBuf2 -->ByteBuf buf = PooledUnsafeDirectByteBuf

NettyBackedChannelBuffer：

1     private ByteBuf buffer;2 3     public NettyBackedChannelBuffer(ByteBuf buffer) {4         Assert.notNull(buffer, "buffer == null");5         this.buffer = buffer;6     }

最终的buffer：

1 NettyBackedChannelBuffer2 -->ByteBuf buffer = SimpleLeakAwareByteBuf3   -->ByteBuf buf = PooledUnsafeDirectByteBuf

 

二、获取NettyChannel

之后从获取io.netty.channel实例，然后包装在NettyChannel中。

1 NettyChannel channel = NettyChannel.getOrAddChannel(ch, url, handler);

1     private static final ConcurrentMap
     
       channelMap = new ConcurrentHashMap
      
       (); 2     private final Channel channel; 3  4     private NettyChannel(Channel channel, URL url, ChannelHandler handler) { 5         super(url, handler); 6         if (channel == null) { 7             throw new IllegalArgumentException("netty channel == null;"); 8         } 9         this.channel = channel;10     }11 12     static NettyChannel getOrAddChannel(Channel ch, URL url, ChannelHandler handler) {13         if (ch == null) {14             return null;15         }16         NettyChannel ret = channelMap.get(ch);17         if (ret == null) {18             NettyChannel nettyChannel = new NettyChannel(ch, url, handler);19             if (ch.isActive()) {20                 ret = channelMap.putIfAbsent(ch, nettyChannel);21             }22             if (ret == null) {23                 ret = nettyChannel;24             }25         }26         return ret;27     }
      
     

首先从缓存ConcurrentMap<Channel, NettyChannel> channelMap中获取key=io.netty.channel的NettyChannel，有则返回，没有则新建并返回。

最终获取到的NettyChannel实例如下：

1 -->Channel channel = NioSocketChannel2 -->ChannelHandler handler = NettyClient3 -->URL url = dubbo://10.10.10.10:20880/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-consumer&check=false&codec=dubbo&default.client=netty4&default.server=netty4&dubbo=2.0.0&generic=false&heartbeat=60000&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&pid=2204&register.ip=10.10.10.10&remote.timestamp=1514958356359&side=consumer&timeout=6000000&timestamp=1514959413199

 

三、进行编码

1 codec.encode(channel, buffer, msg)

这里的codec是：

1 Codec2 codec = 2 DubboCountCodec3 -->DubboCodec codec = new DubboCodec()

DubboCountCodec

1     private DubboCodec codec = new DubboCodec();2 3     public void encode(Channel channel, ChannelBuffer buffer, Object msg) throws IOException {4         codec.encode(channel, buffer, msg);5     }

入参：

• channel：上述的NettyChannel对象
• buffer：上述的NettyBackedChannelBuffer对象
• msg：Request对象，其属性如下：
• 1 long mId = 0 2 String mVersion = "2.0.0" 3 boolean mTwoWay = true 4 boolean mEvent = false 5 boolean mBroken = false 6 Object mData = RpcInvocation对象 7 -->String methodName = "sayHello" 8 -->Class
         [] parameterTypes = [java.lang.String] 9 -->Object[] arguments = ["world"]10 -->Map
         
           attachments = {11      "path" -> "com.alibaba.dubbo.demo.DemoService"12      "interface" -> "com.alibaba.dubbo.demo.DemoService"13      "version" -> "0.0.0"14      "timeout" -> "6000000"15 }16 -->Invoker
           invoker = DubboInvoker对象
         

之后调用DubboCodec.encode(Channel channel, ChannelBuffer buffer, Object msg)，该方法位于其父类ExchangeCodec中。

1     public void encode(Channel channel, ChannelBuffer buffer, Object msg) throws IOException { 2         if (msg instanceof Request) { 3             encodeRequest(channel, buffer, (Request) msg); 4         } else if (msg instanceof Response) { 5             encodeResponse(channel, buffer, (Response) msg); 6         } else { 7             super.encode(channel, buffer, msg); 8         } 9     }10 11     protected void encodeRequest(Channel channel, ChannelBuffer buffer, Request req) throws IOException {12         Serialization serialization = getSerialization(channel);13         // header.14         byte[] header = new byte[HEADER_LENGTH];15         // set magic number.16         Bytes.short2bytes(MAGIC, header);17 18         // set request and serialization flag.19         header[2] = (byte) (FLAG_REQUEST | serialization.getContentTypeId());20 21         if (req.isTwoWay()) header[2] |= FLAG_TWOWAY;22         if (req.isEvent()) header[2] |= FLAG_EVENT;23 24         // set request id.25         Bytes.long2bytes(req.getId(), header, 4);26 27         // encode request data.28         int savedWriteIndex = buffer.writerIndex();29         buffer.writerIndex(savedWriteIndex + HEADER_LENGTH);//设置writerIndex为0+16，先输入请求体的字节30         ChannelBufferOutputStream bos = new ChannelBufferOutputStream(buffer);31         ObjectOutput out = serialization.serialize(channel.getUrl(), bos);32         if (req.isEvent()) {33             encodeEventData(channel, out, req.getData());34         } else {35             encodeRequestData(channel, out, req.getData());36         }37         out.flushBuffer();38         bos.flush();39         bos.close();40         int len = bos.writtenBytes();41         checkPayload(channel, len);42         Bytes.int2bytes(len, header, 12);43 44         // write45         buffer.writerIndex(savedWriteIndex);46         buffer.writeBytes(header); // write header.47         buffer.writerIndex(savedWriteIndex + HEADER_LENGTH + len);48     }

1 首先使用spi获取序列化协议

1 Serialization serialization = getSerialization(channel);

getSerialization位于ExchangeCodec的父类AbstractCodec中。

1     protected Serialization getSerialization(Channel channel) {2         return CodecSupport.getSerialization(channel.getUrl());3     }

1     public static Serialization getSerialization(URL url) {2         return ExtensionLoader.getExtensionLoader(Serialization.class).getExtension(3                 url.getParameter("serialization", "hessian2"));4     }

最终获取到的Serialization serialization = Hessian2Serialization对象：

1 public class Hessian2Serialization implements Serialization { 2     public static final byte ID = 2; 3  4     public byte getContentTypeId() { 5         return ID; 6     } 7  8     public String getContentType() { 9         return "x-application/hessian2";10     }11 12     public ObjectOutput serialize(URL url, OutputStream out) throws IOException {13         return new Hessian2ObjectOutput(out);14     }15 16     public ObjectInput deserialize(URL url, InputStream is) throws IOException {17         return new Hessian2ObjectInput(is);18     }19 }

注意：hessian2序列化方式的id是2，该序列化方式ID会写在协议头里传给服务端，服务端根据序列化方式ID获取对应的序列化方式来反序列化请求体。

2 创建16字节header字节数组

1 byte[] header = new byte[16];

然后填充第1~2个字节为魔数；填充第3个字节为requestFlag、序列化方式ID（这里是2）、twowayFlag或eventFlag；填充第5~12个字节为requestID（long==64bit==8byte）

1         // set magic number. 2         Bytes.short2bytes(MAGIC, header); 3  4         // set request and serialization flag. 5         header[2] = (byte) (FLAG_REQUEST | serialization.getContentTypeId()); 6  7         if (req.isTwoWay()) header[2] |= FLAG_TWOWAY; 8         if (req.isEvent()) header[2] |= FLAG_EVENT; 9 10         // set request id.11         Bytes.long2bytes(req.getId(), header, 4);

3 序列化请求体

首先设置buffer的writerIndex：

1         int savedWriteIndex = buffer.writerIndex();2         buffer.writerIndex(savedWriteIndex + HEADER_LENGTH);//设置writerIndex为0+16，先输入请求体的字节

首先存储了buffer当前的writeIndex（可写位置），从该位置开始到“该位置+15”这一段我们会写入header字节数组（例如，[0,15]），从“该位置+16”开始向后写入请求体字节数组（例如，[16, x)）。

然后就是设置buffer的writerIndex为当前位置+16，因为接下来我们要先序列化请求体，然后将请求体写入buffer，最后才会将header写入buffer。

序列化请求体：

1         ChannelBufferOutputStream bos = new ChannelBufferOutputStream(buffer); 2         ObjectOutput out = serialization.serialize(channel.getUrl(), bos); 3         if (req.isEvent()) { 4             encodeEventData(channel, out, req.getData()); 5         } else { 6             encodeRequestData(channel, out, req.getData()); 7         } 8         out.flushBuffer(); 9         bos.flush();10         bos.close();

首先新建一个ChannelBufferOutputStream对象（该对象继承了java.io.OutputStream抽象类）：

1     private final ChannelBuffer buffer; 2     private final int startIndex; 3  4     public ChannelBufferOutputStream(ChannelBuffer buffer) { 5         if (buffer == null) { 6             throw new NullPointerException("buffer"); 7         } 8         this.buffer = buffer; 9         startIndex = buffer.writerIndex();10     }

buffer为上述的NettyBackedChannelBuffer对象；startIndex == 16

然后获取ObjectOutput对象：

1     public ObjectOutput serialize(URL url, OutputStream out) throws IOException {2         return new Hessian2ObjectOutput(out);3     }

1     private final Hessian2Output mH2o;2 3     public Hessian2ObjectOutput(OutputStream os) {4         mH2o = new Hessian2Output(os);5         mH2o.setSerializerFactory(Hessian2SerializerFactory.SERIALIZER_FACTORY);6     }

1     public final static int SIZE = 4096;2     private final byte[] _buffer = new byte[SIZE];3     protected OutputStream _os;4 5     public Hessian2Output(OutputStream os) {6         _os = os;7     }

最终得到的ObjectOutput对象：

1 Hessian2ObjectOutput2 -->Hessian2Output mH2o3    -->byte[] _buffer = new byte[4096]4    -->OutputStream _os = 上述的ChannelBufferOutputStream对象5    -->SerializerFactory _serializerFactory = Hessian2SerializerFactory实例

最后执行DubboCodec.encodeRequestData(Channel channel, ObjectOutput out, Object data)，该方法是真正的进行请求体序列化的地方。

1     @Override 2     protected void encodeRequestData(Channel channel, ObjectOutput out, Object data) throws IOException { 3         RpcInvocation inv = (RpcInvocation) data; 4  5         out.writeUTF(inv.getAttachment(Constants.DUBBO_VERSION_KEY, DUBBO_VERSION)); 6         out.writeUTF(inv.getAttachment(Constants.PATH_KEY)); 7         out.writeUTF(inv.getAttachment(Constants.VERSION_KEY)); 8  9         out.writeUTF(inv.getMethodName());10         out.writeUTF(ReflectUtils.getDesc(inv.getParameterTypes()));11         Object[] args = inv.getArguments();12         if (args != null)13             for (int i = 0; i < args.length; i++) {14                 out.writeObject(encodeInvocationArgument(channel, inv, i));15             }16         out.writeObject(inv.getAttachments());17     }

其中，channel是上述的NettyChannel实例；out是上述的Hessian2ObjectOutput实例；data是Request对象中的data属性

1 Object mData = RpcInvocation对象 2 -->String methodName = "sayHello" 3 -->Class
     [] parameterTypes = [java.lang.String] 4 -->Object[] arguments = ["world"] 5 -->Map
     
       attachments = { 6      "path" -> "com.alibaba.dubbo.demo.DemoService" 7      "interface" -> "com.alibaba.dubbo.demo.DemoService" 8      "version" -> "0.0.0" 9      "timeout" -> "6000000"10 }11 -->Invoker
       invoker = DubboInvoker对象
     

从DubboCodec.encodeRequestData方法中，我们可以看到只会序列化Request请求体中的RpcInvocation对象的：

• methodName：方法名
• parameterTypes：参数类型
• arguments：参数值
• attachments：附加参数

其中附加参数中的"dubbo"、"path"、"version"还会单独使用out.writeUTF进行序列化。

首先来看一下：

1 Hessian2ObjectOutput.writeUTF(String v)2 -->Hessian2Output.writeString(String value)3    -->printString(String v, int strOffset, int length)

通过这个方法，我们将传入的v存储在ObjectOutput对象的byte[] _buffer = new byte[4096]数组中。

1     Hessian2Output： 2     /** 3      * Writes any object to the output stream. 4      */ 5     public void writeObject(Object object) 6             throws IOException { 7         if (object == null) { 8             writeNull(); 9             return;10         }11 12         Serializer serializer = findSerializerFactory().getSerializer(object.getClass());13         serializer.writeObject(object, this);14     }15 16     public final SerializerFactory findSerializerFactory() {17         SerializerFactory factory = _serializerFactory;18         if (factory == null)19             _serializerFactory = factory = new SerializerFactory();20         return factory;21     }22 23     SerializerFactory：24     private static HashMap _staticSerializerMap;25     private HashMap _cachedSerializerMap;26     /**27      * Returns the serializer for a class.28      * @param cl the class of the object that needs to be serialized.29      * @return a serializer object for the serialization.30      */31     public Serializer getSerializer(Class cl)32             throws HessianProtocolException {33         Serializer serializer;34 35         serializer = (Serializer) _staticSerializerMap.get(cl);36         if (serializer != null)37             return serializer;38 39         if (_cachedSerializerMap != null) {40             synchronized (_cachedSerializerMap) {41                 serializer = (Serializer) _cachedSerializerMap.get(cl);42             }43 44             if (serializer != null)45                 return serializer;46         }47         48         ......49 50         if (serializer != null) {51 52         } 53         .......54         else if (Map.class.isAssignableFrom(cl)) {55             if (_mapSerializer == null)56                 _mapSerializer = new MapSerializer();57 58             serializer = _mapSerializer;59         } 60         ......61         if (serializer == null)62             serializer = getDefaultSerializer(cl);63 64         if (_cachedSerializerMap == null)65             _cachedSerializerMap = new HashMap(8);66 67         synchronized (_cachedSerializerMap) {68             _cachedSerializerMap.put(cl, serializer);69         }70 71         return serializer;72     }

 out.writeObject(Object object)：

首先获取_serializerFactory工厂，这里是Hessian2SerializerFactory实例。其getSerializer(Class cl)方法位于其父类SerializerFactory中：获取序列化器的逻辑是：首先从_staticSerializerMap中获取相关类型的序列化器（_staticSerializerMap中启动时就缓存好一堆类型的序列化器：具体见com.alibaba.com.caucho.hessian.io.SerializerFactory），如果有返回，否则从_cachedSerializerMap缓存中获取相关的类加载器，如果没有，根据类型先创建序列化器（new MapSerializer()，当然还有getDefaultSerializer(cl)来兜底），最后放入缓存_cachedSerializerMap中。最后返回创建好的类加载器。

最后调用MapSerializer.writeObject(Object obj, AbstractHessianOutput out)进行序列化。

DubboCodec.encodeRequestData执行完毕之后，我们将所有的信息写入了ObjectOutput对象的byte[] _buffer = new byte[4096]数组中。

注意：

• 如果在将数据写入到_buffer的过程中，字节量超出了4096，会先执行一把Hessian2ObjectOutput.flushBuffer()将_buffer中的数据拷贝到PooledUnsafeDirectByteBuf中，之后再往_buffer中写入字节

最后执行Hessian2ObjectOutput.flushBuffer()

1 Hessian2ObjectOutput 2     public void flushBuffer() throws IOException { 3         mH2o.flushBuffer(); 4     } 5  6 Hessian2Output 7     public final void flushBuffer() 8             throws IOException { 9         int offset = _offset;10 11         if (!_isStreaming && offset > 0) {12             _offset = 0;13             _os.write(_buffer, 0, offset);14         } else if (_isStreaming && offset > 3) {15             int len = offset - 3;16             _buffer[0] = 'p';17             _buffer[1] = (byte) (len >> 8);18             _buffer[2] = (byte) len;19             _offset = 3;20             _os.write(_buffer, 0, offset);21         }22     }

此处执行ChannelBufferOutputStream.write(byte[] b, int off, int len)

1     @Override2     public void write(byte[] b, int off, int len) throws IOException {3         if (len == 0) {4             return;5         }6         buffer.writeBytes(b, off, len);7     }

1 ChannelBuffer: 2     /** 3      * Transfers the specified source array's data to this buffer starting at 4      * the current {
    @code writerIndex} and increases the {
    @code writerIndex} by 5      * the number of the transferred bytes (= {
    @code length}). 6      * 7      * @param index  the first index of the source 8      * @param length the number of bytes to transfer 9      */10     void writeBytes(byte[] src, int index, int length);

就是将ObjectOutput对象的byte[] _buffer = new byte[4096]数组中的数据转移到buf中。（具体方法见：unsafe.copyMemory( srcBase, long srcOffset,  destBase, long destOffset,long bytes)）

1 NettyBackedChannelBuffer2 -->ByteBuf buffer = SimpleLeakAwareByteBuf3   -->ByteBuf buf = PooledUnsafeDirectByteBuf

4 将header写入buffer

1         int len = bos.writtenBytes();//计算请求体长度2         checkPayload(channel, len);3         Bytes.int2bytes(len, header, 12);//将请求体长度写入header的第13~16个字节（int=4byte）4 5         // write6         buffer.writerIndex(savedWriteIndex);//设置buffer的writerIndex为该次写入的开始位置7         buffer.writeBytes(header); // 将header数组写入buffer8         buffer.writerIndex(savedWriteIndex + HEADER_LENGTH + len);//设置buffer的writerIndex，为下一次写入做准备

到此为止，整个编码就结束了。之后存储了<header><body>数据的ByteBuf由netty自己来进行网络传输。

来看一下请求编码的byte[] header的最终结构:

• 1~2 byte：魔数
• 3 byte：requestFlag、序列化方式ID、twowayFlag或eventFlag
• 5~12 byte ：requestID
• 13~16：请求体长度

这里有一个小插曲：

1     protected static void checkPayload(Channel channel, long size) throws IOException { 2         int payload = Constants.DEFAULT_PAYLOAD; 3         if (channel != null && channel.getUrl() != null) { 4             payload = channel.getUrl().getParameter(Constants.PAYLOAD_KEY, Constants.DEFAULT_PAYLOAD);//8M 5         } 6         if (payload > 0 && size > payload) { 7             ExceedPayloadLimitException e = new ExceedPayloadLimitException("Data length too large: " + size + ", max payload: " + payload + ", channel: " + channel); 8             logger.error(e); 9             throw e;10         }11     }

 dubbo限制了如果传输的请求体长度大于8M，将会直接抛出异常。