简易RPC框架:基于 netty 的协议编解码

概述

《简易RPC框架:需求与设计》这篇文章中已经给出了协议的具体细节,协议类型为二进制协议,如下:

1
2
3
4
5
6
7
8
9
------------------------------------------------------------------------
| magic (2bytes) | version (1byte) | type (1byte) | reserved (7bits) |
------------------------------------------------------------------------
| status (1byte) | id (8bytes) | body length (4bytes) |
------------------------------------------------------------------------
| |
| body ($body_length bytes) |
| |
------------------------------------------------------------------------

协议的解码我们称为 decode,编码我们成为 encode,下文我们将直接使用 decode 和 encode 术语。

decode 的本质就是讲接收到的一串二进制报文,转化为具体的消息对象,在 Java 中,就是将这串二进制报文所包含的信息,用某种类型的对象存储起来。
encode 则是将存储了信息的对象,转化为具有相同含义的一串二进制报文,然后网络收发模块再将报文发出去。

无论是 rpc 客户端还是服务端,都需要有一个 decode 和 encode 的逻辑。

消息类型

rpc 客户端与服务端之间的通信,需要通过发送不同类型的消息来实现,例如:client 向 server 端发送的消息,可能是请求消息,可能是心跳消息,可能是认证消息,而 server 向 client 发送的消息,一般就是响应消息。
利用 Java 中的枚举类型,可以将消息类型进行如下定义:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
/**
* 消息类型
*
* @author beanlam
* @version 1.0
*/
public enum MessageType {

REQUEST((byte) 0x01), HEARTBEAT((byte) 0x02), CHECKIN((byte) 0x03), RESPONSE(
(byte) 0x04), UNKNOWN((byte) 0xFF);

private byte code;

MessageType(byte code) {
this.code = code;
}

public static MessageType valueOf(byte code) {
for (MessageType instance : values()) {
if (instance.code == code) {
return instance;
}
}
return UNKNOWN;
}

public byte getCode() {
return code;
}
}

在这个类中设计了 valueOf 方法,方便进行具体的 byte 字节与具体的消息枚举类型之间的映射和转换。

调用状态设计

client 主动发起的一次 rpc 调用,要么成功,要么失败,server 端有责任告知 client 此次调用的结果,client 也有责任去感知调用失败的原因,因为不一定是 server 端造成的失败,可能是因为 client 端在对消息进行预处理的时候,例如序列化,就已经出错了,这种错误也应该作为一次调用的调用结果返回给 client 调用者。因此引入一个调用状态,与消息类型一样,它也借助了 Java 语言里的枚举类型来实现,并实现了方便的 valueOf 方法:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
/**
* 调用状态
*
* @author beanlam
* @version 1.0
*/
public enum InvocationStatus {

OK((byte) 0x01), CLIENT_TIMEOUT((byte) 0x02), SERVER_TIMEOUT(
(byte) 0x03), BAD_REQUEST((byte) 0x04), BAD_RESPONSE(
(byte) 0x05), SERVICE_NOT_FOUND((byte) 0x06), SERVER_SERIALIZATION_ERROR(
(byte) 0x07), CLIENT_SERIALIZATION_ERROR((byte) 0x08), CLIENT_CANCELED(
(byte) 0x09), SERVER_BUSY((byte) 0x0A), CLIENT_BUSY(
(byte) 0x0B), SERIALIZATION_ERROR((byte) 0x0C), INTERNAL_ERROR(
(byte) 0x0D), SERVER_METHOD_INVOKE_ERROR((byte) 0x0E), UNKNOWN((byte) 0xFF);

private byte code;

InvocationStatus(byte code) {
this.code = code;
}

public static InvocationStatus valueOf(byte code) {
for (InvocationStatus instance : values()) {
if (code == instance.code) {
return instance;
}
}

return UNKNOWN;
}

public byte getCode() {
return code;
}
}

消息实体设计

我们将 client 往 server 端发送的统一称为 rpc 请求消息,一个请求对应着一个响应,因此在 client 和 server 端间流动的信息大体上其实就只有两种,即要么是请求,要么是响应。我们将会定义两个类,分别是 RpcRequest 和 RpcResponse 来代表请求消息和响应消息。
另外由于无论是请求消息还是响应消息,它们都有一些共同的属性,例如说“调用上下文ID”,或者消息类型。因此会再定义一个 RpcMessage 类,作为父类。

RpcMessage

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
/**
* rpc消息
*
* @author beanlam
* @version 1.0
*/
public class RpcMessage {

private MessageType type;
private long contextId;

private Object data;

public long getContextId() {
return this.contextId;
}

public void setContextId(long id) {
this.contextId = id;
}

public Object getData() {
return this.data;
}

public void setData(Object data) {
this.data = data;
}

public void setType(byte code) {
this.type = MessageType.valueOf(code);
}

public MessageType getType() {
return this.type;
}

public void setType(MessageType type) {
this.type = type;
}

@Override
public String toString() {
return "[messageType=" + type.name() + ", contextId=" + contextId + ", data="
+ data + "]";
}
}

RpcRequest

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import java.util.concurrent.atomic.AtomicLong;

/**
* rpc请求消息
*
* @author beanlam
* @version 1.0
*/
public class RpcRequest extends RpcMessage {

private static final AtomicLong ID_GENERATOR = new AtomicLong(0);

public RpcRequest() {
this(ID_GENERATOR.incrementAndGet());
}

public RpcRequest(long contextId) {
setContextId(contextId);
setType(MessageType.REQUEST);
}
}

RpcResponse

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
/**
*
* rpc响应消息
*
* @author beanlam
* @version 1.0
*/
public class RpcResponse extends RpcMessage {

private InvocationStatus status = InvocationStatus.OK;

public RpcResponse(long contextId) {
setContextId(contextId);
setType(MessageType.RESPONSE);
}

public InvocationStatus getStatus() {
return this.status;
}

public void setStatus(InvocationStatus status) {
this.status = status;
}

@Override
public String toString() {
return "RpcResponse[contextId=" + getContextId() + ", status=" + status.name() + "]";
}
}

netty 编解码介绍

netty 是一个 NIO 框架,应该这么说,netty 是一个有良好设计思想的 NIO 框架。一个 NIO 框架必备的要素就是 reactor 线程模型,目前有一些比较优秀而且开源的小型 NIO 框架,例如分库分表中间件 mycat 实现的一个简易 NIO 框架,可以在这里看到。

netty 的主要特点有:微内核设计、责任链模式的业务逻辑处理、内存和资源泄露的检测等。其中编解码在 netty 中,都被设计成责任链上的一个一个 Handler。
decode 对于 netty 来说,它提供了 ByteToMessageDecoder,它也提供了 MessageToByteEncoder。
借助 netty 来实现协议编解码,实际上就是去在这两个handler里面实现编解码的逻辑。

decode

在实现 decode 逻辑时需要注意的一个问题是,由于二进制报文是在网络上发送的,因此一个完整的报文可能经过多个分组来发送的,什么意思呢,就是当有报文进来后,要确认报文是否完整,decode逻辑代码不能假设收到的报文就是一个完整报文,一般称这为“TCP半包问题”。同样,报文是连着报文发送的,意味着decode代码逻辑还要负责在一长串二进制序列中,分割出一个一个独立的报文,这称之为“TCP粘包问题”。
netty 本身有提供一些方便的 decoder handler 来处理 TCP 半包和粘包的问题。不过一般情况下我们不会直接去用它,因为我们的协议比较简单,自己在代码里处理一下就可以了。
完整的 decode 代码逻辑如下所示:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
import cn.com.agree.ats.rpc.message.*;
import cn.com.agree.ats.util.logfacade.AbstractPuppetLoggerFactory;
import cn.com.agree.ats.util.logfacade.IPuppetLogger;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;

import java.util.List;

/**
* 协议解码器
*
* @author beanlam
* @version 1.0
*/
public class ProtocolDecoder extends ByteToMessageDecoder {

private static final IPuppetLogger logger = AbstractPuppetLoggerFactory
.getInstance(ProtocolDecoder.class);

private boolean magicChecked = false;

@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> list)
throws Exception {

if (!magicChecked) {
if (in.readableBytes() < ProtocolMetaData.MAGIC_LENGTH_IN_BYTES) {
return;
}
magicChecked = true;

if (!(in.getShort(in.readerIndex()) == ProtocolMetaData.MAGIC)) {
logger.warn(
"illegal data received without correct magic number, channel will be close");
ctx.close();
magicChecked = false; //this line of code makes no any sense, but it's good for a warning
return;
}
}

if (in.readableBytes() < ProtocolMetaData.HEADER_LENGTH_IN_BYTES) {
return;
}

int bodyLength = in
.getInt(in.readerIndex() + ProtocolMetaData.BODY_LENGTH_OFFSET);
if (in.readableBytes() < bodyLength + ProtocolMetaData.HEADER_LENGTH_IN_BYTES) {
return;
}
magicChecked = false;// so far the whole packet was received

in.readShort(); // skip the magic
in.readByte(); // dont care about the protocol version so far
byte type = in.readByte();
byte status = in.readByte();
long contextId = in.readLong();

byte[] body = new byte[in.readInt()];
in.readBytes(body);

RpcMessage message = null;
MessageType messageType = MessageType.valueOf(type);

if (messageType == MessageType.RESPONSE) {
message = new RpcResponse(contextId);
((RpcResponse) message).setStatus(InvocationStatus.valueOf(status));
} else {
message = new RpcRequest(contextId);
}
message.setType(messageType);
message.setData(body);
list.add(message);
}
}

可以看到,我们解决半包问题的时候,是判断有没有收到我们期望收到的报文,如果没有,直接在 decode 方法里面 return,等有更多的报文被收到的时候,netty 会自动帮我们调起 decode 方法。而我们解决粘包问题的思路也很清晰,那就是一次只处理一个报文,不去动后面的报文内容。

还需要注意的是,在 netty 中,对于 ByteBuf 的 get 是不会消费掉报文的,而 read 是会消费掉报文的。当不确定报文是否收完整的时候,我们都是用 get开头的方法去试探性地验证报文是否接收完全,当确定报文接收完全后,我们才用 read 开头的方法去消费这段报文。

encode

直接贴代码,参考前文提到的协议格式阅读以下代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
/**
*
* 协议编码器
*
* @author beanlam
* @version 1.0
*/
public class ProtocolEncoder extends MessageToByteEncoder<RpcMessage> {

@Override
protected void encode(ChannelHandlerContext ctx, RpcMessage rpcMessage, ByteBuf out)
throws Exception {
byte status;
byte[] data = (byte[]) rpcMessage.getData();

if (rpcMessage instanceof RpcRequest) {
RpcRequest request = (RpcRequest) rpcMessage;
status = InvocationStatus.OK.getCode();
} else {
RpcResponse response = (RpcResponse) rpcMessage;
status = response.getStatus().getCode();
}

out.writeShort(ProtocolMetaData.MAGIC);
out.writeByte(ProtocolMetaData.VERSION);
out.writeByte(rpcMessage.getType().getCode());
out.writeByte(status);
out.writeLong(rpcMessage.getContextId());
out.writeInt(data.length);
out.writeBytes(data);
}
}

beanlam wechat
欢迎订阅我的个人公众号
壕无人性