作者:vivo 互联网服务器团队-Sun wen
一、背景
二、Dubbo编解码
/**
* Init bootstrap
*
* @throws Throwable
*/
protected void doOpen() throws Throwable {
bootstrap = new Bootstrap();
// ...
bootstrap.handler(new ChannelInitializer<SocketChannel>() {
protected void initChannel(SocketChannel ch) throws Exception {
// ...
ch.pipeline()
.addLast("decoder", adapter.getDecoder())
.addLast("encoder", adapter.getEncoder())
.addLast("client-idle-handler", new IdleStateHandler(heartbeatInterval, 0, 0, MILLISECONDS))
.addLast("handler", nettyClientHandler);
// ...
}
});
}
/**
* Init and start netty server
*
* @throws Throwable
*/
protected void doOpen() throws Throwable {
bootstrap = new ServerBootstrap();
// ...
bootstrap.group(bossGroup, workerGroup)
.channel(NettyEventLoopFactory.serverSocketChannelClass())
.option(ChannelOption.SO_REUSEADDR, Boolean.TRUE)
.childOption(ChannelOption.TCP_NODELAY, Boolean.TRUE)
.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
.childHandler(new ChannelInitializer<SocketChannel>() {
protected void initChannel(SocketChannel ch) throws Exception {
// ...
ch.pipeline()
.addLast("decoder", adapter.getDecoder())
.addLast("encoder", adapter.getEncoder())
.addLast("server-idle-handler", new IdleStateHandler(0, 0, idleTimeout, MILLISECONDS))
.addLast("handler", nettyServerHandler);
}
});
// ...
}
ChannelInboundHandler
...
NettyCodecAdapter#getEncoder()
->NettyCodecAdapter$InternalEncoder#encode
->DubboCountCodec#encode
->DubboCodec#encode
->ExchangeCodec#encode
->ExchangeCodec#encodeRequest
DubboCountCodec类实际引用的是DubboCodec,因DubboCodec继承于ExchangeCodec,并未重写encode方法,所以实际代码跳转会直接进入ExchangeCodec#encode方法
NettyCodecAdapter#getDecoder()
->NettyCodecAdapter$InternalDecoder#decode
->DubboCountCodec#decode
->DubboCodec#decode
->ExchangeCodec#decode
->DubboCodec#decodeBody
...
MultiMessageHandler#received
->HeartbeatHadnler#received
->AllChannelHandler#received
...
ChannelEventRunnable#run
->DecodeHandler#received
->DecodeHandler#decode
->DecodeableRpcResult#decode
解码链路相对复杂,过程中做了两次解码,在一次DubboCodec#decodeBody内,并未实际解码channel的数据,而是构建成DecodeableRpcResult对象,然后在业务处理的Handler里通过异步线程进行实际解码。
NettyCodecAdapter#getDecoder()
->NettyCodecAdapter$InternalDecoder#decode
->DubboCountCodec#decode
->DubboCodec#decode
->ExchangeCodec#decode
->DubboCodec#decodeBody
...
MultiMessageHandler#received
->HeartbeatHadnler#received
->AllChannelHandler#received
...
ChannelEventRunnable#run
->DecodeHandler#received
->DecodeHandler#decode
->DecodeableRpcInvocation#decode
提供端解码链路与消费端的类似,区别在于实际解码对象不一样,DecodeableRpcResult 替换成 DecodeableRpcInvocation。
体现了Dubbo代码里的良好设计,抽象处理链路,屏蔽处理细节,流程清晰可复用。
NettyCodecAdapter#getEncoder()
->NettyCodecAdapter$InternalEncoder#encode
->DubboCountCodec#encode
->DubboCodec#encode
->ExchangeCodec#encode
->ExchangeCodec#encodeResponse
与消费方发送消息链路一致,区别在于最后一步区分Request和Response,进行不同内容编码
2byte:magic,类似java字节码文件里的魔数,用来标识是否是dubbo协议的数据包。
1byte:消息标志位,5位序列化id,1位心跳还是正常请求,1位双向还是单向,1位请求还是响应;
1byte:响应状态,具体类型见com.alibaba.dubbo.remoting.exchange.Response; 8byte:消息ID,每一个请求的唯一识别id;
4byte:消息体body长度。
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);
// body length
Bytes.int2bytes(len, header, 12);
// write
buffer.writerIndex(savedWriteIndex);
buffer.writeBytes(header); // write header.
buffer.writerIndex(savedWriteIndex + HEADER_LENGTH + len);
}
三、Hessian2
@Overrideprotected void encodeRequestData(Channel channel, ObjectOutput out, Object data, String version) throws IOException { RpcInvocation inv = (RpcInvocation) data; out.writeUTF(version); // https://github.com/apache/dubbo/issues/6138 String serviceName = inv.getAttachment(INTERFACE_KEY); if (serviceName == null) { serviceName = inv.getAttachment(PATH_KEY); } out.writeUTF(serviceName); out.writeUTF(inv.getAttachment(VERSION_KEY)); out.writeUTF(inv.getMethodName()); out.writeUTF(inv.getParameterTypesDesc()); Object[] args = inv.getArguments(); if (args != null) { for (int i = 0; i < args.length; i++) { out.writeObject(encodeInvocationArgument(channel, inv, i)); } } out.writeAttachments(inv.getObjectAttachments());}
protected void encodeRequest(Channel channel, ChannelBuffer buffer, Request req) throws IOException {
Serialization serialization = getSerialization(channel);
// ...
ObjectOutput out = serialization.serialize(channel.getUrl(), bos);
if (req.isEvent()) {
encodeEventData(channel, out, req.getData());
} else {
encodeRequestData(channel, out, req.getData(), req.getVersion());
}
// ...
}
public static Serialization getSerialization(URL url) {
return ExtensionLoader.getExtensionLoader(Serialization.class).getExtension(
url.getParameter(Constants.SERIALIZATION_KEY, Constants.DEFAULT_REMOTING_SERIALIZATION));
}
@Override
public ObjectOutput serialize(URL url, OutputStream out) throws IOException {
return new Hessian2ObjectOutput(out);
}
@Override
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...
try {
DecodeableRpcResult result;
if (channel.getUrl().getParameter(DECODE_IN_IO_THREAD_KEY, 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;
} catch (Throwable t) {
// ...
}
return res;
} else {
// decode request...
return req;
}
}
if (channel.getUrl().getParameter(DECODE_IN_IO_THREAD_KEY, DEFAULT_DECODE_IN_IO_THREAD)) {
inv = new DecodeableRpcInvocation(channel, req, is, proto);
inv.decode();
} else {
inv = new DecodeableRpcInvocation(channel, req,
new UnsafeByteArrayInputStream(readMessageData(is)), proto);
}
@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:
// ...
case DubboCodec.RESPONSE_VALUE_WITH_ATTACHMENTS:
handleValue(in);
handleAttachment(in);
break;
case DubboCodec.RESPONSE_WITH_EXCEPTION_WITH_ATTACHMENTS:
// ...
default:
throw new IOException("Unknown result flag, expect '0' '1' '2' '3' '4' '5', but received: " + flag);
}
// ...
return this;
}
private void handleValue(ObjectInput in) throws IOException {
try {
Type[] returnTypes;
if (invocation instanceof RpcInvocation) {
returnTypes = ((RpcInvocation) invocation).getReturnTypes();
} else {
returnTypes = RpcUtils.getReturnTypes(invocation);
}
Object value = null;
if (ArrayUtils.isEmpty(returnTypes)) {
// This almost never happens?
value = in.readObject();
} else if (returnTypes.length == 1) {
value = in.readObject((Class<?>) returnTypes[0]);
} else {
value = in.readObject((Class<?>) returnTypes[0], returnTypes[1]);
}
setValue(value);
} catch (ClassNotFoundException e) {
rethrow(e);
}
}
四、常见问题
public JavaSerializer(Class cl, ClassLoader loader) {
introspectWriteReplace(cl, loader);
// ...
List fields = new ArrayList();
fields.addAll(primitiveFields);
fields.addAll(compoundFields);
Collections.reverse(fields);
// ...
}
五、写在最后
END
猜你喜欢
vivo互联网技术
vivo移动互联网是基于vivo 智能手机所建立的完整移动互联网生态圈,围绕vivo大数据运营,打造包括应用、游戏、资讯、品牌、电商、内容、金融、搜索的全方位服务生态,满足海量用户的多样化需求。
点一下,代码无 Bug