阐述Dubbo服务提供方的解码原理

本文主要是介绍阐述Dubbo服务提供方的解码原理，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

1 解码概述

在Dubbo服务消费方（客户端）和服务提供方（服务端）进行网络通信时，服务提供方会通过socket把需要发送的内容序列化为二进制流后发出。接着二进制流通过网络流向服务提供方。服务提供方接收到该请求后会解析该请求包，对接收到的数据进行反序列化后对请求进行处理。

服务提供方接收到请求后解析请求包（即Dubbo协议的内容）的过程即服务提供方的解码过程。

2 解码原理解析

2.1 解码入口

服务提供方解析请求包最终是借助于 InternalDecoder 的 decode 方法来实现的。具体实现代码如下所示。

protected void decode(ChannelHandlerContext ctx, ByteBuf input, List<Object> out) throws Exception {ChannelBuffer message = new NettyBackedChannelBuffer(input);NettyChannel channel = NettyChannel.getOrAddChannel(ctx.channel(), url, handler);// decode object.do {int saveReaderIndex = message.readerIndex();Object msg = codec.decode(channel, message);if (msg == Codec2.DecodeResult.NEED_MORE_INPUT) {message.readerIndex(saveReaderIndex);break;} else {//is it possible to go here ?if (saveReaderIndex == message.readerIndex()) {throw new IOException("Decode without read data.");}if (msg != null) {out.add(msg);}}} while (message.readable());
}

2.2 解码实现细节

在 InternalDecoder 的 decode 方法内部再调用了具体解码实现来进行解码。如 ExchangeCodec 的 decode 方法，实现如下所示。

public Object decode(Channel channel, ChannelBuffer buffer) throws IOException {// 获取dubbo协议帧的字节数int readable = buffer.readableBytes();// 将dubbo协议头读取到数组headerbyte[] header = new byte[Math.min(readable, HEADER_LENGTH)];buffer.readBytes(header);// 解析dubbo协议帧数据部分return decode(channel, buffer, readable, header);
}protected Object decode(Channel channel, ChannelBuffer buffer, int readable, byte[] header) throws IOException {// check magic number.if (readable > 0 && header[0] != MAGIC_HIGH|| readable > 1 && header[1] != MAGIC_LOW) {int length = header.length;if (header.length < readable) {header = Bytes.copyOf(header, readable);buffer.readBytes(header, length, readable - length);}for (int i = 1; i < header.length - 1; i++) {if (header[i] == MAGIC_HIGH && header[i + 1] == MAGIC_LOW) {buffer.readerIndex(buffer.readerIndex() - header.length + i);header = Bytes.copyOf(header, i);break;}}return super.decode(channel, buffer, readable, header);}// check length.// （1）dubbo协议帧实际字节数小于协议头长度，说明该协议帧不是一个完整的协议帧if (readable < HEADER_LENGTH) {return DecodeResult.NEED_MORE_INPUT;}// get data length.int len = Bytes.bytes2int(header, 12);// When receiving response, how to exceed the length, then directly construct a response to the client.// see more detail from https://github.com/apache/dubbo/issues/7021.Object obj = finishRespWhenOverPayload(channel, len, header);if (null != obj) {return obj;}// dubbo协议帧理论上的字节数int tt = len + HEADER_LENGTH;// （2）dubbo协议帧实际字节数小于理论上的字节数，说明该协议帧不是一个完整的协议帧if (readable < tt) {return DecodeResult.NEED_MORE_INPUT;}// limit input stream.ChannelBufferInputStream is = new ChannelBufferInputStream(buffer, len);try {return decodeBody(channel, is, header);} finally {if (is.available() > 0) {try {if (logger.isWarnEnabled()) {logger.warn(TRANSPORT_SKIP_UNUSED_STREAM, "", "", "Skip input stream " + is.available());}StreamUtils.skipUnusedStream(is);} catch (IOException e) {logger.warn(TRANSPORT_SKIP_UNUSED_STREAM, "", "", e.getMessage(), e);}}}
}

其中将dubbo协议头读取到header数组的实现如下所示。

    @Overridepublic void readBytes(byte[] dst) {readBytes(dst, 0, dst.length);}

对dubbo协议body进行解析则是调用了 decodeBody 方法，实现如下所示。

protected Object decodeBody(Channel channel, InputStream is, byte[] header) throws IOException {byte flag = header[2], proto = (byte) (flag & SERIALIZATION_MASK);// get request id.long id = Bytes.bytes2long(header, 4);if ((flag & FLAG_REQUEST) == 0) {// decode response.Response res = new Response(id);if ((flag & FLAG_EVENT) != 0) {res.setEvent(true);}// get status.byte status = header[3];res.setStatus(status);try {if (status == Response.OK) {Object data;if (res.isEvent()) {byte[] eventPayload = CodecSupport.getPayload(is);if (CodecSupport.isHeartBeat(eventPayload, proto)) {// heart beat response data is always null;data = null;} else {data = decodeEventData(channel, CodecSupport.deserialize(channel.getUrl(), new ByteArrayInputStream(eventPayload), proto), eventPayload);}} else {data = decodeResponseData(channel, CodecSupport.deserialize(channel.getUrl(), is, proto), getRequestData(channel, res, id));}res.setResult(data);} else {res.setErrorMessage(CodecSupport.deserialize(channel.getUrl(), is, proto).readUTF());}} catch (Throwable t) {res.setStatus(Response.CLIENT_ERROR);res.setErrorMessage(StringUtils.toString(t));}return res;} else {// decode request.Request req;try {Object data;if ((flag & FLAG_EVENT) != 0) {byte[] eventPayload = CodecSupport.getPayload(is);if (CodecSupport.isHeartBeat(eventPayload, proto)) {// heart beat response data is always null;req = new HeartBeatRequest(id);((HeartBeatRequest) req).setProto(proto);data = null;} else {req = new Request(id);data = decodeEventData(channel, CodecSupport.deserialize(channel.getUrl(), new ByteArrayInputStream(eventPayload), proto), eventPayload);}req.setEvent(true);} else {req = new Request(id);data = decodeRequestData(channel, CodecSupport.deserialize(channel.getUrl(), is, proto));}req.setData(data);} catch (Throwable t) {// bad requestreq = new Request(id);req.setBroken(true);req.setData(t);}req.setVersion(Version.getProtocolVersion());req.setTwoWay((flag & FLAG_TWOWAY) != 0);return req;}
}

这篇关于阐述Dubbo服务提供方的解码原理的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---