Dubbo框架设计源码解读五(消息编解码器)
write by valuewithTime, 2020-11-25 23:00引言
Netty服务端是基于经典的bootstrap，事件，worker, 编解码器，消息处理器的实现。netty处理器实际为一个共享的SimpleChannelHandler， 所有操作委托内部通道处理器DecodeHandler。DecodeHandler在接受消息后，解码相应的消息，将交由内部的HeaderExchangeHandler处理。HeaderExchangeHandler， 所有的操作委托给内部的ExchangeHander，实际为DubboProtocol的中ExchangeHandleAdater。
消费者调用服务提供者实际上发送的一个Invocation消息，服务端接受到消息，根据Invoker上下文，从Dubbo协议的Exporter容器中获取对应的Invoker，调用相关服务，将调用结果返回给消费者。
netty客户端也没有多少新鲜的动心，编解码器，Netty客户端处理器NettyClientHandler。NettyClientHandler实际为一个共享的ChannelDuplexHandler，所有操作委托内部通道处理器DecodeHandler。DecodeHandler在接受消息后，解码相应的消息，将交由内部的HeaderExchangeHandler处理。HeanderExchangeHandler，HeanderExchangeHandler所有的操作委托给内部的ExchangeHander，实际为DubboProtocol的中ExchangeHandleAdater。一个请求，主要包括请求Id，版本，及数据及Invocation。服务响应，主要包含消息id，消息版本，状态，相应结果，及错误信息，如果有的话。
这是上一篇所讲内容，今天我们来看一下编解码。
目录
概要框架设计
dubbo框架主要包括序列化，消息层，传输层，协议层。序列化主要是请求消息和响应消息的序列化，比如基于Javad的ObjectOut/InputStream序列化、基于JSON的序列化。消息层提供消费者调用服务请求消息、服务提供方处理 结果响应消息的编解码；传输层主要建立消费者和服务者的通信通道，传输服务请求响应数据，比如基于Netty和Mina的，默认为Netty；协议层首先是基于相关协议将服提供者，和消费者通过export暴露出去，即注册器Registry中，消费者通过Registry订阅响应的服务提供者，消费者发现有服务 提供者，则与服务提供者建立连接，注册协议有基于Zookeeper，Redis等，在注册协议中还有一个注册器目录服务，用于提供消费者和服务者列表，及根据负载均衡策略选择服务者。服务提供者接受的消费者的服务请求后，根据相关协议，调用相应的Invoker服务。 消费者和服务者的RPC调用协议，实际在DubboProtocol中，协议首先导出服务，消费者发送RPC请求，调用Exporter服务容器中的Invoker。
源码分析
Dubbo框架设计源码解读第一篇（服务和引用bean初始化）
应用协议
注册器协议
Dubbo协议
Dubbo框架设计源码解读三（Dubbo协议，服务导出，引用）
数据传输器Transport
Dubbo框架设计源码解读四(Dubbo基于Netty的传输器Transport)
消息编解码
在传输器Transport的基于Netty的服务端和客户端中，都一个编解码适配器，今天我们来看一下，如何进行编解码的。
NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this);
ChannelPipeline pipeline = Channels.pipeline();
/*int idleTimeout = getIdleTimeout();
if (idleTimeout > 10000) {
pipeline.addLast("timer", new IdleStateHandler(timer, idleTimeout / 1000, 0, 0));
}*/
pipeline.addLast("decoder", adapter.getDecoder());
pipeline.addLast("encoder", adapter.getEncoder());
pipeline.addLast("handler", nettyHandler);
//NettyCodecAdapter
/**
* NettyCodecAdapter.
*/
final class NettyCodecAdapter {
private final ChannelHandler encoder = new InternalEncoder();
private final ChannelHandler decoder = new InternalDecoder();
private final Codec2 codec;
private final URL url;
private final org.apache.dubbo.remoting.ChannelHandler handler;
public NettyCodecAdapter(Codec2 codec, URL url, org.apache.dubbo.remoting.ChannelHandler handler) {
this.codec = codec;
this.url = url;
this.handler = handler;
}
public ChannelHandler getEncoder() {
return encoder;
}
public ChannelHandler getDecoder() {
return decoder;
}
/**
*
*/
private class InternalEncoder extends MessageToByteEncoder {
@Override
protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception {
org.apache.dubbo.remoting.buffer.ChannelBuffer buffer = new NettyBackedChannelBuffer(out);
Channel ch = ctx.channel();
NettyChannel channel = NettyChannel.getOrAddChannel(ch, url, handler);
try {
codec.encode(channel, buffer, msg);
} finally {
NettyChannel.removeChannelIfDisconnected(ch);
}
}
}
/**
*
*/
private class InternalDecoder extends ByteToMessageDecoder {
@Override
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);
Object msg;
int saveReaderIndex;
try {
// decode object.
do {
saveReaderIndex = message.readerIndex();
try {
msg = codec.decode(channel, message);
} catch (IOException e) {
throw e;
}
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());
} finally {
NettyChannel.removeChannelIfDisconnected(ctx.channel());
}
}
}
}
NettyCodecAdapter为编解码器的适配，内部编码器实际为ByteToMessageDecoder， 内部的编码操作委托给内部编解码器，根据SPI机制，实际为DubboCodec；内部解码器实际为ByteToMessageDecoder，解码操作委托给DubboCodec。
下面我们分别来看一下内部编解码器
消息编码器
/**
*
*/
private class InternalEncoder extends MessageToByteEncoder {
@Override
protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception {
org.apache.dubbo.remoting.buffer.ChannelBuffer buffer = new NettyBackedChannelBuffer(out);
Channel ch = ctx.channel();
NettyChannel channel = NettyChannel.getOrAddChannel(ch, url, handler);
try {
codec.encode(channel, buffer, msg);
} finally {
NettyChannel.removeChannelIfDisconnected(ch);
}
}
}
//ExchangeCodec
**
* ExchangeCodec.
*
*
*
*/
public class ExchangeCodec extends TelnetCodec {
// header length. 头部长度
protected static final int HEADER_LENGTH = 16;
// magic header. 魔数头部
protected static final short MAGIC = (short) 0xdabb;
protected static final byte MAGIC_HIGH = Bytes.short2bytes(MAGIC)[0];
protected static final byte MAGIC_LOW = Bytes.short2bytes(MAGIC)[1];
// message flag. 消息标志
protected static final byte FLAG_REQUEST = (byte) 0x80;
protected static final byte FLAG_TWOWAY = (byte) 0x40;
protected static final byte FLAG_EVENT = (byte) 0x20;
/**
* 序列化掩码
*/
protected static final int SERIALIZATION_MASK = 0x1f;
private static final Logger logger = LoggerFactory.getLogger(ExchangeCodec.class);
public Short getMagicCode() {
return MAGIC;
}
@Override
public void encode(Channel channel, ChannelBuffer buffer, Object msg) throws IOException {
if (msg instanceof Request) {
encodeRequest(channel, buffer, (Request) msg);
} else if (msg instanceof Response) {
encodeResponse(channel, buffer, (Response) msg);
} else {
super.encode(channel, buffer, msg);
}
}
...
/**
* @param channel
* @param buffer
* @param req
* @throws IOException
*/
protected void encodeRequest(Channel channel, ChannelBuffer buffer, Request req) throws IOException {
Serialization serialization = getSerialization(channel);
// header.
byte[] header = new byte[HEADER_LENGTH];
// set magic number.
Bytes.short2bytes(MAGIC, header);
// set request and serialization flag.
header[2] = (byte) (FLAG_REQUEST | serialization.getContentTypeId());
if (req.isTwoWay()) {
header[2] |= FLAG_TWOWAY;
}
if (req.isEvent()) {
header[2] |= FLAG_EVENT;
}
// set request id.
Bytes.long2bytes(req.getId(), header, 4);
// encode request data.
int savedWriteIndex = buffer.writerIndex();
buffer.writerIndex(savedWriteIndex + HEADER_LENGTH);
ChannelBufferOutputStream bos = new ChannelBufferOutputStream(buffer);
ObjectOutput out = serialization.serialize(channel.getUrl(), bos);
if (req.isEvent()) {
encodeEventData(channel, out, req.getData());
} else {
encodeRequestData(channel, out, req.getData(), req.getVersion());
}
out.flushBuffer();
if (out instanceof Cleanable) {
((Cleanable) out).cleanup();
}
bos.flush();
bos.close();
int len = bos.writtenBytes();
checkPayload(channel, len);
Bytes.int2bytes(len, header, 12);
// write
buffer.writerIndex(savedWriteIndex);
buffer.writeBytes(header); // write header.
buffer.writerIndex(savedWriteIndex + HEADER_LENGTH + len);
}
/**
* @param channel
* @param buffer
* @param res
* @throws IOException
*/
protected void encodeResponse(Channel channel, ChannelBuffer buffer, Response res) throws IOException {
int savedWriteIndex = buffer.writerIndex();
try {
Serialization serialization = getSerialization(channel);
// header.
byte[] header = new byte[HEADER_LENGTH];
// set magic number.
Bytes.short2bytes(MAGIC, header);
// set request and serialization flag.
header[2] = serialization.getContentTypeId();
if (res.isHeartbeat()) {
header[2] |= FLAG_EVENT;
}
// set response status.
byte status = res.getStatus();
header[3] = status;
// set request id.
Bytes.long2bytes(res.getId(), header, 4);
buffer.writerIndex(savedWriteIndex + HEADER_LENGTH);
ChannelBufferOutputStream bos = new ChannelBufferOutputStream(buffer);
ObjectOutput out = serialization.serialize(channel.getUrl(), bos);
// encode response data or error message.
if (status == Response.OK) {
if (res.isHeartbeat()) {
encodeHeartbeatData(channel, out, res.getResult());
} else {
encodeResponseData(channel, out, res.getResult(), res.getVersion());
}
} else {
out.writeUTF(res.getErrorMessage());
}
out.flushBuffer();
if (out instanceof Cleanable) {
((Cleanable) out).cleanup();
}
bos.flush();
bos.close();
int len = bos.writtenBytes();
checkPayload(channel, len);
Bytes.int2bytes(len, header, 12);
// write
buffer.writerIndex(savedWriteIndex);
buffer.writeBytes(header); // write header.
buffer.writerIndex(savedWriteIndex + HEADER_LENGTH + len);
} catch (Throwable t) {
// clear buffer
buffer.writerIndex(savedWriteIndex);
// send error message to Consumer, otherwise, Consumer will wait till timeout.
if (!res.isEvent() && res.getStatus() != Response.BAD_RESPONSE) {
Response r = new Response(res.getId(), res.getVersion());
r.setStatus(Response.BAD_RESPONSE);
if (t instanceof ExceedPayloadLimitException) {
logger.warn(t.getMessage(), t);
try {
r.setErrorMessage(t.getMessage());
channel.send(r);
return;
} catch (RemotingException e) {
logger.warn("Failed to send bad_response info back: " + t.getMessage() + ", cause: " + e.getMessage(), e);
}
} else {
// FIXME log error message in Codec and handle in caught() of IoHanndler?
logger.warn("Fail to encode response: " + res + ", send bad_response info instead, cause: " + t.getMessage(), t);
try {
r.setErrorMessage("Failed to send response: " + res + ", cause: " + StringUtils.toString(t));
channel.send(r);
return;
} catch (RemotingException e) {
logger.warn("Failed to send bad_response info back: " + res + ", cause: " + e.getMessage(), e);
}
}
}
// Rethrow exception
if (t instanceof IOException) {
throw (IOException) t;
} else if (t instanceof RuntimeException) {
throw (RuntimeException) t;
} else if (t instanceof Error) {
throw (Error) t;
} else {
throw new RuntimeException(t.getMessage(), t);
}
}
}
@Override
protected void encodeData(ObjectOutput out, Object data) throws IOException {
encodeRequestData(out, data);
}
private void encodeEventData(ObjectOutput out, Object data) throws IOException {
out.writeObject(data);
}
@Deprecated
protected void encodeHeartbeatData(ObjectOutput out, Object data) throws IOException {
encodeEventData(out, data);
}
protected void encodeRequestData(ObjectOutput out, Object data) throws IOException {
out.writeObject(data);
}
protected void encodeResponseData(ObjectOutput out, Object data) throws IOException {
out.writeObject(data);
}
}
来看编码请求和响应的具体实现
//DubboCodec
/**
* @param channel
* @param out
* @param data
* @throws IOException
*/
@Override
protected void encodeRequestData(Channel channel, ObjectOutput out, Object data) throws IOException {
encodeRequestData(channel, out, data, DUBBO_VERSION);
}
/**
* @param channel
* @param out
* @param data
* @throws IOException
*/
@Override
protected void encodeResponseData(Channel channel, ObjectOutput out, Object data) throws IOException {
encodeResponseData(channel, out, data, DUBBO_VERSION);
}
@Override
protected void encodeRequestData(Channel channel, ObjectOutput out, Object data, String version) throws IOException {
//RPC调用
RpcInvocation inv = (RpcInvocation) data;
out.writeUTF(version);
out.writeUTF(inv.getAttachment(Constants.PATH_KEY));
out.writeUTF(inv.getAttachment(Constants.VERSION_KEY));
out.writeUTF(inv.getMethodName());
out.writeUTF(ReflectUtils.getDesc(inv.getParameterTypes()));
Object[] args = inv.getArguments();
if (args != null) {
for (int i = 0; i < args.length; i++) {
out.writeObject(encodeInvocationArgument(channel, inv, i));
}
}
out.writeObject(RpcUtils.getNecessaryAttachments(inv));
}
@Override
protected void encodeResponseData(Channel channel, ObjectOutput out, Object data, String version) throws IOException {
Result result = (Result) data;
// currently, the version value in Response records the version of Request
boolean attach = Version.isSupportResponseAttachment(version);
Throwable th = result.getException();
if (th == null) {
Object ret = result.getValue();
if (ret == null) {
out.writeByte(attach ? RESPONSE_NULL_VALUE_WITH_ATTACHMENTS : RESPONSE_NULL_VALUE);
} else {
out.writeByte(attach ? RESPONSE_VALUE_WITH_ATTACHMENTS : RESPONSE_VALUE);
out.writeObject(ret);
}
} else {
out.writeByte(attach ? RESPONSE_WITH_EXCEPTION_WITH_ATTACHMENTS : RESPONSE_WITH_EXCEPTION);
out.writeObject(th);
}
if (attach) {
// returns current version of Response to consumer side.
result.getAttachments().put(Constants.DUBBO_VERSION_KEY, Version.getProtocolVersion());
out.writeObject(result.getAttachments());
}
}
来看编码请求数据
@Override
protected void encodeRequestData(Channel channel, ObjectOutput out, Object data, String version) throws IOException {
//RPC调用
RpcInvocation inv = (RpcInvocation) data;
out.writeUTF(version);
out.writeUTF(inv.getAttachment(Constants.PATH_KEY));
out.writeUTF(inv.getAttachment(Constants.VERSION_KEY));
out.writeUTF(inv.getMethodName());
out.writeUTF(ReflectUtils.getDesc(inv.getParameterTypes()));
Object[] args = inv.getArguments();
if (args != null) {
for (int i = 0; i < args.length; i++) {
out.writeObject(encodeInvocationArgument(channel, inv, i));
}
}
out.writeObject(RpcUtils.getNecessaryAttachments(inv));
}
//ReflectUtils
**
* ReflectUtils
*/
public final class ReflectUtils {
/**
* void(V).
*/
public static final char JVM_VOID = 'V';
/**
* boolean(Z).
*/
public static final char JVM_BOOLEAN = 'Z';
/**
* byte(B).
*/
public static final char JVM_BYTE = 'B';
/**
* char(C).
*/
public static final char JVM_CHAR = 'C';
/**
* double(D).
*/
public static final char JVM_DOUBLE = 'D';
/**
* float(F).
*/
public static final char JVM_FLOAT = 'F';
/**
* int(I).
*/
public static final char JVM_INT = 'I';
/**
* long(J).
*/
public static final char JVM_LONG = 'J';
/**
* short(S).
*/
public static final char JVM_SHORT = 'S';
public static final Class<?>[] EMPTY_CLASS_ARRAY = new Class<?>[0];
public static final String JAVA_IDENT_REGEX = "(?:[_$a-zA-Z][_$a-zA-Z0-9]*)";
public static final String JAVA_NAME_REGEX = "(?:" + JAVA_IDENT_REGEX + "(?:\\." + JAVA_IDENT_REGEX + ")*)";
/**
*
*/
public static final String CLASS_DESC = "(?:L" + JAVA_IDENT_REGEX + "(?:\\/" + JAVA_IDENT_REGEX + ")*;)";
/**
*
*/
public static final String ARRAY_DESC = "(?:\\[+(?:(?:[VZBCDFIJS])|" + CLASS_DESC + "))";
/**
*
*/
public static final String DESC_REGEX = "(?:(?:[VZBCDFIJS])|" + CLASS_DESC + "|" + ARRAY_DESC + ")";
public static final Pattern DESC_PATTERN = Pattern.compile(DESC_REGEX);
public static final String METHOD_DESC_REGEX = "(?:(" + JAVA_IDENT_REGEX + ")?\\((" + DESC_REGEX + "*)\\)(" + DESC_REGEX + ")?)";
public static final Pattern METHOD_DESC_PATTERN = Pattern.compile(METHOD_DESC_REGEX);
public static final Pattern GETTER_METHOD_DESC_PATTERN = Pattern.compile("get([A-Z][_a-zA-Z0-9]*)\\(\\)(" + DESC_REGEX + ")");
public static final Pattern SETTER_METHOD_DESC_PATTERN = Pattern.compile("set([A-Z][_a-zA-Z0-9]*)\\((" + DESC_REGEX + ")\\)V");
public static final Pattern IS_HAS_CAN_METHOD_DESC_PATTERN = Pattern.compile("(?:is|has|can)([A-Z][_a-zA-Z0-9]*)\\(\\)Z");
private static final ConcurrentMap<String, Class<?>> DESC_CLASS_CACHE = new ConcurrentHashMap<String, Class<?>>();
private static final ConcurrentMap<String, Class<?>> NAME_CLASS_CACHE = new ConcurrentHashMap<String, Class<?>>();
private static final ConcurrentMap<String, Method> Signature_METHODS_CACHE = new ConcurrentHashMap<String, Method>();
private ReflectUtils() {
}
...
/**
* get class array desc.
* [int.class, boolean[].class, Object.class] => "I[ZLjava/lang/Object;"
*
* @param cs class array.
* @return desc.
* @throws NotFoundException
*/
public static String getDesc(final Class<?>[] cs) {
if (cs.length == 0) {
return "";
}
StringBuilder sb = new StringBuilder(64);
for (Class<?> c : cs) {
sb.append(getDesc(c));
}
return sb.toString();
}
/**
* get class desc.
* boolean[].class => "[Z"
* Object.class => "Ljava/lang/Object;"
*
* @param c class.
* @return desc.
* @throws NotFoundException
*/
public static String getDesc(Class<?> c) {
StringBuilder ret = new StringBuilder();
while (c.isArray()) {
ret.append('[');
c = c.getComponentType();
}
if (c.isPrimitive()) {
String t = c.getName();
if ("void".equals(t)) {
ret.append(JVM_VOID);
} else if ("boolean".equals(t)) {
ret.append(JVM_BOOLEAN);
} else if ("byte".equals(t)) {
ret.append(JVM_BYTE);
} else if ("char".equals(t)) {
ret.append(JVM_CHAR);
} else if ("double".equals(t)) {
ret.append(JVM_DOUBLE);
} else if ("float".equals(t)) {
ret.append(JVM_FLOAT);
} else if ("int".equals(t)) {
ret.append(JVM_INT);
} else if ("long".equals(t)) {
ret.append(JVM_LONG);
} else if ("short".equals(t)) {
ret.append(JVM_SHORT);
}
} else {
ret.append('L');
ret.append(c.getName().replace('.', '/'));
ret.append(';');
}
return ret.toString();
}
}
从上面可以看出，编码器编码主要是针对请求Request和响应Response，编码首先写头部，针对请求主要有魔数、序列化标志，请求id，然后是请求数据,主要有RpcInvocation的服务URL，版本信息，服务方法，方法参数类型，方法参数类型通过ReflectUtils进行编码，最后还有方法参数；针对响应，首先写头部，魔数，请求id，然后是响应数据。需要注意在编码请求和响应的时候，有一部分是Attament，这部分是可以扩展的地方。
来看一下序列化器
//CodecSupport
/**
* @param url
* @return
*/
public static Serialization getSerialization(URL url) {
return ExtensionLoader.getExtensionLoader(Serialization.class).getExtension(
url.getParameter(Constants.SERIALIZATION_KEY, Constants.DEFAULT_REMOTING_SERIALIZATION));
}
序列化器，有FastJsonSerialization，JavaSerialization, 默认为JavaSerialization，我们来看JavaSerialization的实现。
/**
*
*/
public class JavaSerialization implements Serialization {
@Override
public byte getContentTypeId() {
return 3;
}
@Override
public String getContentType() {
return "x-application/java";
}
@Override
public ObjectOutput serialize(URL url, OutputStream out) throws IOException {
return new JavaObjectOutput(out);
}
@Override
public ObjectInput deserialize(URL url, InputStream is) throws IOException {
return new JavaObjectInput(is);
}
}
先来看序列化
//JavaObjectOutput
/**
* Java Object output.
*/
public class JavaObjectOutput extends NativeJavaObjectOutput {
public JavaObjectOutput(OutputStream os) throws IOException {
super(new ObjectOutputStream(os));
}
public JavaObjectOutput(OutputStream os, boolean compact) throws IOException {
super(compact ? new CompactedObjectOutputStream(os) : new ObjectOutputStream(os));
}
@Override
public void writeUTF(String v) throws IOException {
if (v == null) {
getObjectOutputStream().writeInt(-1);
} else {
getObjectOutputStream().writeInt(v.length());
getObjectOutputStream().writeUTF(v);
}
}
@Override
public void writeObject(Object obj) throws IOException {
if (obj == null) {
getObjectOutputStream().writeByte(0);
} else {
getObjectOutputStream().writeByte(1);
getObjectOutputStream().writeObject(obj);
}
}
@Override
public void flushBuffer() throws IOException {
getObjectOutputStream().flush();
}
}
反序列化 //JavaObjectInput
/**
* Java Object input.
*/
public class JavaObjectInput extends NativeJavaObjectInput {
public final static int MAX_BYTE_ARRAY_LENGTH = 8 * 1024 * 1024;
public JavaObjectInput(InputStream is) throws IOException {
super(new ObjectInputStream(is));
}
public JavaObjectInput(InputStream is, boolean compacted) throws IOException {
super(compacted ? new CompactedObjectInputStream(is) : new ObjectInputStream(is));
}
@Override
public byte[] readBytes() throws IOException {
int len = getObjectInputStream().readInt();
if (len < 0) {
return null;
}
if (len == 0) {
return new byte[0];
}
if (len > MAX_BYTE_ARRAY_LENGTH) {
throw new IOException("Byte array length too large. " + len);
}
byte[] b = new byte[len];
getObjectInputStream().readFully(b);
return b;
}
@Override
public String readUTF() throws IOException {
int len = getObjectInputStream().readInt();
if (len < 0) {
return null;
}
return getObjectInputStream().readUTF();
}
@Override
public Object readObject() throws IOException, ClassNotFoundException {
byte b = getObjectInputStream().readByte();
if (b == 0) {
return null;
}
return getObjectInputStream().readObject();
}
@Override
@SuppressWarnings("unchecked")
public <T> T readObject(Class<T> cls) throws IOException,
ClassNotFoundException {
return (T) readObject();
}
@Override
@SuppressWarnings("unchecked")
public <T> T readObject(Class<T> cls, Type type) throws IOException, ClassNotFoundException {
return (T) readObject();
}
}
从上面可看出，JavaSerialization的序列化和反序列化实际是基于ObjectOutput/InputStream。
我们再来看一下，解码器，对编码的反相关操作，我们简单过一下
消息解码器
/**
*
*/
private class InternalDecoder extends ByteToMessageDecoder {
@Override
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);
Object msg;
int saveReaderIndex;
try {
// decode object.
do {
saveReaderIndex = message.readerIndex();
try {
msg = codec.decode(channel, message);
} catch (IOException e) {
throw e;
}
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());
} finally {
NettyChannel.removeChannelIfDisconnected(ctx.channel());
}
}
}
//ExchangeCodec
@Override
public Object decode(Channel channel, ChannelBuffer buffer) throws IOException {
int readable = buffer.readableBytes();
byte[] header = new byte[Math.min(readable, HEADER_LENGTH)];
buffer.readBytes(header);
return decode(channel, buffer, readable, header);
}
@Override
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.
if (readable < HEADER_LENGTH) {
return DecodeResult.NEED_MORE_INPUT;
}
// get data length.
int len = Bytes.bytes2int(header, 12);
checkPayload(channel, len);
int tt = len + HEADER_LENGTH;
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("Skip input stream " + is.available());
}
StreamUtils.skipUnusedStream(is);
} catch (IOException e) {
logger.warn(e.getMessage(), e);
}
}
}
}
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(Response.HEARTBEAT_EVENT);
}
// get status.
byte status = header[3];
res.setStatus(status);
try {
ObjectInput in = CodecSupport.deserialize(channel.getUrl(), is, proto);
if (status == Response.OK) {
Object data;
if (res.isHeartbeat()) {
data = decodeHeartbeatData(channel, in);
} else if (res.isEvent()) {
data = decodeEventData(channel, in);
} else {
data = decodeResponseData(channel, in, getRequestData(id));
}
res.setResult(data);
} else {
res.setErrorMessage(in.readUTF());
}
} catch (Throwable t) {
res.setStatus(Response.CLIENT_ERROR);
res.setErrorMessage(StringUtils.toString(t));
}
return res;
} else {
// decode request.
Request req = new Request(id);
req.setVersion(Version.getProtocolVersion());
req.setTwoWay((flag & FLAG_TWOWAY) != 0);
if ((flag & FLAG_EVENT) != 0) {
req.setEvent(Request.HEARTBEAT_EVENT);
}
try {
ObjectInput in = CodecSupport.deserialize(channel.getUrl(), is, proto);
Object data;
if (req.isHeartbeat()) {
data = decodeHeartbeatData(channel, in);
} else if (req.isEvent()) {
data = decodeEventData(channel, in);
} else {
data = decodeRequestData(channel, in);
}
req.setData(data);
} catch (Throwable t) {
// bad request
req.setBroken(true);
req.setData(t);
}
return req;
}
}
/**
* @param channel
* @param in
* @return
* @throws IOException
*/
protected Object decodeEventData(Channel channel, ObjectInput in) throws IOException {
try {
return in.readObject();
} catch (ClassNotFoundException e) {
throw new IOException(StringUtils.toString("Read object failed.", e));
}
}
/**
* @param channel
* @param in
* @return
* @throws IOException
*/
@Deprecated
protected Object decodeHeartbeatData(Channel channel, ObjectInput in) throws IOException {
try {
return in.readObject();
} catch (ClassNotFoundException e) {
throw new IOException(StringUtils.toString("Read object failed.", e));
}
}
protected Object decodeRequestData(Channel channel, ObjectInput in) throws IOException {
return decodeRequestData(in);
}
protected Object decodeResponseData(Channel channel, ObjectInput in) throws IOException {
return decodeResponseData(in);
}
protected Object decodeResponseData(Channel channel, ObjectInput in, Object requestData) throws IOException {
return decodeResponseData(channel, in);
}
//DubboCodec
@Override
protected Object decodeBody(Channel channel, InputStream is, byte[] header) throws IOException {
byte flag = header[2], proto = (byte) (flag & SERIALIZATION_MASK);
// get request id. 请求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(Response.HEARTBEAT_EVENT);
}
// get status. 状态
byte status = header[3];
res.setStatus(status);
try {
//从输入流中获取对象
ObjectInput in = CodecSupport.deserialize(channel.getUrl(), is, proto);
if (status == Response.OK) {
Object data;
if (res.isHeartbeat()) {
//心跳
data = decodeHeartbeatData(channel, in);
} else if (res.isEvent()) {
//事件
data = decodeEventData(channel, in);
} else {
//解码RPC调用结果
DecodeableRpcResult result;
//IO线程
if (channel.getUrl().getParameter(
Constants.DECODE_IN_IO_THREAD_KEY,
Constants.DEFAULT_DECODE_IN_IO_THREAD)) {
result = new DecodeableRpcResult(channel, res, is,
(Invocation) getRequestData(id), proto);
result.decode();
} else {
result = new DecodeableRpcResult(channel, res,
new UnsafeByteArrayInputStream(readMessageData(is)),
(Invocation) getRequestData(id), proto);
}
data = result;
}
res.setResult(data);
} else {
res.setErrorMessage(in.readUTF());
}
} catch (Throwable t) {
if (log.isWarnEnabled()) {
log.warn("Decode response failed: " + t.getMessage(), t);
}
res.setStatus(Response.CLIENT_ERROR);
res.setErrorMessage(StringUtils.toString(t));
}
return res;
} else {
// decode request.
Request req = new Request(id);
req.setVersion(Version.getProtocolVersion());
req.setTwoWay((flag & FLAG_TWOWAY) != 0);
if ((flag & FLAG_EVENT) != 0) {
req.setEvent(Request.HEARTBEAT_EVENT);
}
try {
Object data;
ObjectInput in = CodecSupport.deserialize(channel.getUrl(), is, proto);
if (req.isHeartbeat()) {
data = decodeHeartbeatData(channel, in);
} else if (req.isEvent()) {
data = decodeEventData(channel, in);
} else {
DecodeableRpcInvocation inv;
if (channel.getUrl().getParameter(
Constants.DECODE_IN_IO_THREAD_KEY,
Constants.DEFAULT_DECODE_IN_IO_THREAD)) {
inv = new DecodeableRpcInvocation(channel, req, is, proto);
inv.decode();
} else {
//解码RPC调用
inv = new DecodeableRpcInvocation(channel, req,
new UnsafeByteArrayInputStream(readMessageData(is)), proto);
}
data = inv;
}
req.setData(data);
} catch (Throwable t) {
if (log.isWarnEnabled()) {
log.warn("Decode request failed: " + t.getMessage(), t);
}
// bad request
req.setBroken(true);
req.setData(t);
}
return req;
}
}
protected Object getRequestData(long id) {
DefaultFuture future = DefaultFuture.getFuture(id);
if (future == null) {
return null;
}
Request req = future.getRequest();
if (req == null) {
return null;
}
return req.getData();
}
再来看一下解码请求
//
public class DecodeableRpcInvocation extends RpcInvocation implements Codec, Decodeable {
private static final Logger log = LoggerFactory.getLogger(DecodeableRpcInvocation.class);
/**
*
*/
private Channel channel;
/**
*
*/
private byte serializationType;
/**
*
*/
private InputStream inputStream;
/**
*
*/
private Request request;
/**
*
*/
private volatile boolean hasDecoded;
public DecodeableRpcInvocation(Channel channel, Request request, InputStream is, byte id) {
Assert.notNull(channel, "channel == null");
Assert.notNull(request, "request == null");
Assert.notNull(is, "inputStream == null");
this.channel = channel;
this.request = request;
this.inputStream = is;
this.serializationType = id;
}
...
@Override
public Object decode(Channel channel, InputStream input) throws IOException {
ObjectInput in = CodecSupport.getSerialization(channel.getUrl(), serializationType)
.deserialize(channel.getUrl(), input);
//dubbo版本
String dubboVersion = in.readUTF();
request.setVersion(dubboVersion);
setAttachment(Constants.DUBBO_VERSION_KEY, dubboVersion);
setAttachment(Constants.PATH_KEY, in.readUTF());
setAttachment(Constants.VERSION_KEY, in.readUTF());
//方法名
setMethodName(in.readUTF());
try {
Object[] args;
Class<?>[] pts;
String desc = in.readUTF();
if (desc.length() == 0) {
pts = DubboCodec.EMPTY_CLASS_ARRAY;
args = DubboCodec.EMPTY_OBJECT_ARRAY;
} else {
//参数类型
pts = ReflectUtils.desc2classArray(desc);
//参数值
args = new Object[pts.length];
for (int i = 0; i < args.length; i++) {
try {
args[i] = in.readObject(pts[i]);
} catch (Exception e) {
if (log.isWarnEnabled()) {
log.warn("Decode argument failed: " + e.getMessage(), e);
}
}
}
}
//设置参数类型
setParameterTypes(pts);
//获取attachment
Map<String, String> map = (Map<String, String>) in.readObject(Map.class);
if (map != null && map.size() > 0) {
Map<String, String> attachment = getAttachments();
if (attachment == null) {
attachment = new HashMap<String, String>();
}
attachment.putAll(map);
setAttachments(attachment);
}
//decode argument ,may be callback ， 解码参数
for (int i = 0; i < args.length; i++) {
args[i] = decodeInvocationArgument(channel, this, pts, i, args[i]);
}
//设置参数
setArguments(args);
} catch (ClassNotFoundException e) {
throw new IOException(StringUtils.toString("Read invocation data failed.", e));
} finally {
if (in instanceof Cleanable) {
((Cleanable) in).cleanup();
}
}
return this;
}
}
再来看一下解码请求响应 //DecodeableRpcResult
public class DecodeableRpcResult extends RpcResult implements Codec, Decodeable {
private static final Logger log = LoggerFactory.getLogger(DecodeableRpcResult.class);
private Channel channel;
private byte serializationType;
private InputStream inputStream;
private Response response;
private Invocation invocation;
private volatile boolean hasDecoded;
...
@Override
public Object decode(Channel channel, InputStream input) throws IOException {
ObjectInput in = CodecSupport.getSerialization(channel.getUrl(), serializationType)
.deserialize(channel.getUrl(), input);
byte flag = in.readByte();
switch (flag) {
case DubboCodec.RESPONSE_NULL_VALUE:
break;
case DubboCodec.RESPONSE_VALUE:
try {
Type[] returnType = RpcUtils.getReturnTypes(invocation);
setValue(returnType == null || returnType.length == 0 ? in.readObject() :
(returnType.length == 1 ? in.readObject((Class<?>) returnType[0])
: in.readObject((Class<?>) returnType[0], returnType[1])));
} catch (ClassNotFoundException e) {
throw new IOException(StringUtils.toString("Read response data failed.", e));
}
break;
case DubboCodec.RESPONSE_WITH_EXCEPTION:
try {
Object obj = in.readObject();
if (obj instanceof Throwable == false) {
throw new IOException("Response data error, expect Throwable, but get " + obj);
}
setException((Throwable) obj);
} catch (ClassNotFoundException e) {
throw new IOException(StringUtils.toString("Read response data failed.", e));
}
break;
case DubboCodec.RESPONSE_NULL_VALUE_WITH_ATTACHMENTS:
try {
setAttachments((Map<String, String>) in.readObject(Map.class));
} catch (ClassNotFoundException e) {
throw new IOException(StringUtils.toString("Read response data failed.", e));
}
break;
case DubboCodec.RESPONSE_VALUE_WITH_ATTACHMENTS:
try {
Type[] returnType = RpcUtils.getReturnTypes(invocation);
setValue(returnType == null || returnType.length == 0 ? in.readObject() :
(returnType.length == 1 ? in.readObject((Class<?>) returnType[0])
: in.readObject((Class<?>) returnType[0], returnType[1])));
setAttachments((Map<String, String>) in.readObject(Map.class));
} catch (ClassNotFoundException e) {
throw new IOException(StringUtils.toString("Read response data failed.", e));
}
break;
case DubboCodec.RESPONSE_WITH_EXCEPTION_WITH_ATTACHMENTS:
try {
Object obj = in.readObject();
if (obj instanceof Throwable == false) {
throw new IOException("Response data error, expect Throwable, but get " + obj);
}
setException((Throwable) obj);
setAttachments((Map<String, String>) in.readObject(Map.class));
} catch (ClassNotFoundException e) {
throw new IOException(StringUtils.toString("Read response data failed.", e));
}
break;
default:
throw new IOException("Unknown result flag, expect '0' '1' '2', get " + flag);
}
if (in instanceof Cleanable) {
((Cleanable) in).cleanup();
}
return this;
}
}
//RpcUtils
/**
* RpcUtils
*/
public class RpcUtils {
...
/**
* TODO why not get return type when initialize Invocation?
* @param invocation
* @return
*/
public static Type[] getReturnTypes(Invocation invocation) {
try {
if (invocation != null && invocation.getInvoker() != null
&& invocation.getInvoker().getUrl() != null
&& !invocation.getMethodName().startsWith("$")) {
String service = invocation.getInvoker().getUrl().getServiceInterface();
if (service != null && service.length() > 0) {
Class<?> invokerInterface = invocation.getInvoker().getInterface();
Class<?> cls = invokerInterface != null ? ReflectUtils.forName(invokerInterface.getClassLoader(), service)
: ReflectUtils.forName(service);
Method method = cls.getMethod(invocation.getMethodName(), invocation.getParameterTypes());
if (method.getReturnType() == void.class) {
return null;
}
Class<?> returnType = method.getReturnType();
Type genericReturnType = method.getGenericReturnType();
if (Future.class.isAssignableFrom(returnType)) {
if (genericReturnType instanceof ParameterizedType) {
Type actualArgType = ((ParameterizedType) genericReturnType).getActualTypeArguments()[0];
if (actualArgType instanceof ParameterizedType) {
returnType = (Class<?>) ((ParameterizedType) actualArgType).getRawType();
genericReturnType = actualArgType;
} else {
returnType = (Class<?>) actualArgType;
genericReturnType = returnType;
}
} else {
returnType = null;
genericReturnType = null;
}
}
return new Type[]{returnType, genericReturnType};
}
}
} catch (Throwable t) {
logger.warn(t.getMessage(), t);
}
return null;
}
}
从上面可以看出，解码消息首先解码消息头部魔数等数据，然后是请求id，如果是消费者请求则解码出消息体，实际为DecodeableRpcInvocation，包括方面名，参数以及参数值；如果是请求响应，解码出返回结果，实际为DecodeableRpcResult，包含服务响应数据。
总结
NettyCodecAdapter为编解码器的适配，内部编码器实际为ByteToMessageDecoder， 内部的编码操作委托给内部编解码器，根据SPI机制，实际为DubboCodec；内部解码器实际为ByteToMessageDecoder，解码操作委托给DubboCodec。
编码器编码主要是针对请求Request和响应Response，编码首先写头部，针对请求主要有魔数、序列化标志，请求id，然后是请求数据,主要有RpcInvocation的服务URL，版本信息，服务方法，方法参数类型，方法参数类型通过ReflectUtils进行编码，最后还有方法参数；针对响应，首先写头部，魔数，请求id，然后是响应数据。需要注意在编码请求和响应的时候，有一部分是Attament，这部分是可以扩展的地方。序列化器，有FastJsonSerialization，JavaSerialization, 默认为JavaSerialization。JavaSerialization的序列化和反序列化实际是基于ObjectOutput/InputStream。
解码消息首先解码消息头部魔数等数据，然后是请求id，如果是消费者请求则解码出消息体，实际为DecodeableRpcInvocation，包括方面名，参数以及参数值；如果是请求响应，解码出返回结果，实际为DecodeableRpcResult，包含服务响应数据。
附
dubbo offical site
dubbo github
dubbo github vt
incubator-dubbo-spring-boot-project github vt