基于netty的通讯协议的设计思考_JAVA_编程开发_程序员俱乐部

# 一、dubbo协议的设计

? 先从业务开始，思考设计协议。

## 1. 从rpc到remote

? dubbo的目标是：一个接口方法的动态实现了远程方法调用，让调用者感觉与本地调用一样。既然是远程调用，涉及到用什么协议把调用数据发过去，以及接收方按这个协议解析出请求，进行处理后，用这个协议再返回结果。

对于rpc要做的事情有：

• - 调用方要做的事情有：产生一个包含调用数据的DubboInvoker，并且要用一个传输工具把这个发过去。
• - 被调用方应该早就做好的事情：把自己所拥有的接口与实现，分别转换为DubboInvoker：service（key:value），存在map中，另外应该启动好了一个传输的服务端来接收DubboInvoker。

那remote-传输工具要做的事情就是：

• - 调用方把DubboInvoker，编码成指定的dubbo协议格式。
• - 被调用方把数据流，转换成DubboInvoker。DubboInvoker就是一个java对象，包含了远程调用的方法参数等信息的类。

## 2. remote的处理

? DubboInvoker只是要传输的RPC业务内容，现在再考虑一下传输。通常我们会给快递的内容装进一个有基本信息的盒子，所以传输的是一个包含了head与body的信息，body就是DubboInvoker。传输可以是：mina、netty、grizzy。就是几家快递公司，不同的是东西会转成byte[]。传输层还有自己的业务内容，比如是不要回执，心跳等，是请求还是回复。这样，内容就分了两个层次，传输本身的与RPC的。另外内容有个编码过程，所以有两层次的编码。

**比如三种传输工具都可以从外部给一个编码方式进去。比如就是dubbo协议吧。你给它们DubboInvoker，它们就会用编码器转换/传输/再反转换：**

class="java" name="code">```java
//mina服务端启动时加载一个编码适配器，带着编码方式。
acceptor.getFilterChain().addLast("codec", new ProtocolCodecFilter(new MinaCodecAdapter(getCodec(), getUrl(), this)));

//Grizzly服务端启动时，也加载一个编码适配器，带着编码方式。
 filterChainBuilder.add(new GrizzlyCodecAdapter(getCodec(), getUrl(), this));
 
//Netty4服务端启动时，也加载一个编码适配器，带着编码方式。
protected void initChannel(NioSocketChannel ch) throws Exception {
NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this);
ch.pipeline()//.addLast("logging",new LoggingHandler(LogLevel.INFO))//for debug
.addLast("decoder", adapter.getDecoder())
.addLast("encoder", adapter.getEncoder())
.addLast("handler", nettyServerHandler);
}
```

## 3. 协议是怎么回事

就是怎么把DubboInvoker包装成一个request的body，并把request的body怎么转成DubboInvoker的过程是核心业务，应该是协议的重要部分。传输工具的客户端与服务器也要有消息交流的需求，比如心跳啊。这样要区分不同类型的消息，就要有一个消息头，里面至少要有类型，如果是心跳，就可能没有body，或者body里是其它对象。

所以，RPC的传输只是传输的内容的一部分，当然也只是协议的一部分。为了满足公共传输部分的要求，所以协议要分成head与body两个层次。

协议的头部是满足多种消息的传输使用，而且对这几个工具是无区别的，可以定义在remote模块的公共包中，不可能放在netty4或者mina中吧。如果头部如果发现类型是DubboInvoker，再按格式解析body，这个与rpc密切相关，只能放在rpc-dubbo模块的包里了。我传输工具只解析头部进行处理，如果是DubboInvoker对象，给你这rpc这层再处理。

传输工具肯定有一个exchangeHandler，而且还持有一个rpcHandler，如果自己能处理的消息，就不用rpcHandler了。

```java
//exchangeHandler.Received():
//处理Request对象，根据类型，特殊时交给内部持有的rpcHandler处理。可以暂时认为是rpcHandler，可能会被包装一下，插入点其它事情。
if (message instanceof Request) {
            // handle request.
            Request request = (Request) message;
            if (request.isEvent()) {
                handlerEvent(channel, request);
            } else {
                if (request.isTwoWay()) {
                    Response response = handleRequest(exchangeChannel, request);//里面用了handler
                    channel.send(response);
                } else {
                    handler.received(exchangeChannel, request.getData());//直接用了handler
                }
            }
        } 

//---------------------------------------------------
//DubboProtocol.java里会有new ExchangeHandlerAdapter() {...}，就是上面的handler。它的received处理的已经是Invocation了。
        @Override
        public void received(Channel channel, Object message) throws RemotingException {
            if (message instanceof Invocation) {
                reply((ExchangeChannel) channel, message);
            } else {
                super.received(channel, message);
            }
        }
```

所以，我认为dubbo协议是对request的head与body的规范，分别用在remote与rpc层。从code类的继承关系，也可以看出是明显的层次，dubboCodec是进一步的处理。

```java
public class DubboCodec extends ExchangeCodec implements Codec2
```

```html
magic：类似java字节码文件里的魔数，用来判断是不是dubbo协议的数据包。魔数是常量0xdabb,用于判断报文的开始。
flag：标志位, 一共8个地址位。低四位用来表示消息体数据用的序列化工具的类型（默认hessian），高四位中，第一位为1表示是request请求，第二位为1表示双向传输（即有返回response），第三位为1表示是心跳ping事件。
status：状态位, 设置请求响应状态，dubbo定义了一些响应的类型。具体类型见
  com.alibaba.dubbo.remoting.exchange.Response
invoke id：消息id, long 类型。每一个请求的唯一识别id（由于采用异步通讯的方式，用来把请求request和返回的response对应上）
body length：消息体 body 长度, int 类型，即记录Body Content有多少个字节。
```

## 4. 序列化

序列化支持：dubbo、hessian2、java、json。只是协议中的相关的类转成byte[]的格式的规范。

```java
//DubboCodec.java的方法中可以看到，序列化是另一回事，从url中配置的。看到header在remote中已经解析好了，给rpc部分直接用。
//可是，应该可以给一个只不过body没解析的Request对象过来吧？？？？
//应该不处理heatbeat与event了吧？？？？？
@Override
protected Object decodeBody(Channel channel, InputStream is, byte[] header) throws IOException {
    byte flag = header[2], proto = (byte) (flag & SERIALIZATION_MASK);
    Serialization s = CodecSupport.getSerialization(channel.getUrl(), proto);
    ...
        
            Request req = new Request(id);
            req.setVersion(Version.getProtocolVersion());
            req.setTwoWay((flag & FLAG_TWOWAY) != 0);
...
            try {
                Object data;
                if (req.isHeartbeat()) {
                    //decodeHeartbeatData已经deprecated了，按说这里也不会出现了。
                    data = decodeHeartbeatData(channel, deserialize(s, channel.getUrl(), is));
                } else if (req.isEvent()) {
                    data = decodeEventData(channel, deserialize(s, channel.getUrl(), is));
                } else {
                    DecodeableRpcInvocation inv;
...//这里是RpcInvocation了，后面给了body里。
                        inv = new DecodeableRpcInvocation(channel, req,
                                new UnsafeByteArrayInputStream(readMessageData(is)), proto);
...
                    data = inv;
                }
                req.setData(data);
```

## 5. 设计特点

? 看到dubbo的协议，很容易想到之前常用的把http当成传输工具的api接口设计。通常在http返回200时，才处理http的内容，里面还有业务的返回码与可能的结果。这也是明显的业务与传输的区分，不过现在流行的restful，需要把http协议当成应用协议，比如返回，你就不应该再自己定义返回码了，用http那套。

传输层的协议特点：

• - 消息类型：很重要，放在了flag中的高四位中。两种业务，一种心跳。
• - 消息id：异步时，返回消息按这个找请求消息，所以也必须。http就没有这个。
• - 序列化类型：如果是统一的，就没必要。这里的可配置的，接受端按这个序列化body中的数据。
• - 消息长度：对于tcp处理粘包，拆包很重要。netty可以有定长，分隔，也有基于长度的方式。
• - status：响应类型，也很重要，可能会不解析body。

RPC业务的部分特点：

• - MethodName：方法名字
• - ParameterTypes：方法参数类型
• - Arguments：参数对象
• - Attachments：附件
• - version：接口版本，与head层的版本可能意思不同。

# 二、 rocketmq的协议设计

## 1. RocketMq的特点

• - 所有的传输都是自己用的，不给外部用。不象dubbo重点是远程RPC，要兼容很多东西。所以选择最常用，单一的方式进行消息传输。
• - 并且它不是RPC传输，所以body部分不是协议重点。
• - RocketMq的网络通信是基于netty4.x实现的，这下传输工具是确定的。
• - 传输什么样都消息都是确定数目与类型的，比如同步主备数据，比如获取namesvr数据。

协议格式 <length> <header length> <header data> <body data>

?                                1                      2                        3                      4

?        1、4个字节的int型数据来存储2、3、4的总长度

?        2、4个字节的int型数据来存储报文头部的字节长度等于3的长度

?        3、存储报文头部的数据

?        4、存储报文体的数据



## 2. 协议头的设计与使用

### 2.1 消息的结构与发送业务

以发送客户端的消息队列消息为例，客户的消息是byte[]类型，但还包含发到哪个队列，哪个主题，时间等属性。

```java
//RemotingCommand.java
//协议头部（传输的消息）设计如下：
private int code;
private LanguageCode language = LanguageCode.JAVA;
private int version = 0;
private int opaque = requestId.getAndIncrement();
private int flag = 0;
private String remark;
private HashMap<String, String> extFields;//customHeader中的内容转成string,string。序列化这里的内容。而CommandCustomHeader由消息的使用方，根据消息的code进行强转类型。
private transient CommandCustomHeader customHeader;//不进行序列化

private SerializeType serializeTypeCurrentRPC = serializeTypeConfigInThisServer;

private transient byte[] body;
```

```java
//向消息队列发消息
public SendResult sendMessage(
    final String addr,
    final String brokerName,
    final Message msg,
    final SendMessageRequestHeader requestHeader,//
    final long timeoutMillis,
    final CommunicationMode communicationMode,
    final SendCallback sendCallback,
    final TopicPublishInfo topicPublishInfo,
    final MQClientInstance instance,
    final int retryTimesWhenSendFailed,
    final SendMessageContext context,
    final DefaultMQProducerImpl producer
) throws RemotingException, MQBrokerException, InterruptedException {
    long beginStartTime = System.currentTimeMillis();
    RemotingCommand request = null;
...
        request = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE, requestHeader);
...
    request.setBody(msg.getBody());//消息生产者的消息，作为通讯消息的body，这个解码由最终用户来定了。
    switch (communicationMode) {
        case ONEWAY:
            this.remotingClient.invokeOneway(addr, request, timeoutMillis);
            return null;
            }
     ...
     }

//-----------------------------------------------------
//RemotingCommand.createRequestCommand时，设计这两个要消息头。
        cmd.setCode(code);//消息的业务功能类型，确定数目的类型。
        cmd.customHeader = customHeader;//与这个类型相关的业务数据对象，与code一一对应。


//SendMessageRequestHeader都是简单的类型，设置customHeader包含：
SendMessageRequestHeader requestHeader = new SendMessageRequestHeader();
requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());
requestHeader.setTopic(msg.getTopic());
requestHeader.setDefaultTopic(this.defaultMQProducer.getCreateTopicKey());
requestHeader.setDefaultTopicQueueNums(this.defaultMQProducer.getDefaultTopicQueueNums());
requestHeader.setQueueId(mq.getQueueId());
requestHeader.setSysFlag(sysFlag);
...

   
```

### 2.2 消息的传输与codec

NettyDecoder继承自netty的长度区分数据包，方法中得到完整数据包后，就转成RemotingCommand。

```java
public class NettyDecoder extends LengthFieldBasedFrameDecoder
    @Override
    public Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
        ByteBuf frame = null;
        try {
            frame = (ByteBuf) super.decode(ctx, in);//用super方法，得到正确拆TCP包后的数据帧
            if (null == frame) {
                return null;
            }
            ByteBuffer byteBuffer = frame.nioBuffer();
            return RemotingCommand.decode(byteBuffer);//具体解析有效果的数据帧，转成RemotingCommand
        } catch (Exception e) {
            ...
        } finally {
           ...
        }
        return null;
    }
    

//RemotingCommand.decode()中，主要是解析头部


//headerDecode有两种序列化方式。
    private static RemotingCommand headerDecode(byte[] headerData, SerializeType type) {
        switch (type) {
            case JSON:
                RemotingCommand resultJson = RemotingSerializable.decode(headerData, RemotingCommand.class);
                resultJson.setSerializeTypeCurrentRPC(type);
                return resultJson;
            case ROCKETMQ:
                RemotingCommand resultRMQ = RocketMQSerializable.rocketMQProtocolDecode(headerData);
                resultRMQ.setSerializeTypeCurrentRPC(type);
                return resultRMQ;
            default:
                break;
        }
        return null;
    }
```

> encode的时候，由使用RemotingCommand方法，进行 extFields与customHeader之间的转换，所以netty解析出来的半成品，要设置setSerializeTypeCurrentRPC，这样使用方可以解析extFields与customHeader。



### 2.3 netty传输工具的使用

netty的协议，主要就是处理业务的，不涉及到其它不相关的方面，比如不涉及心跳。

```java
public void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline()
                        .addLast(defaultEventExecutorGroup, HANDSHAKE_HANDLER_NAME,
                            new HandshakeHandler(TlsSystemConfig.tlsMode))
                        .addLast(defaultEventExecutorGroup,
                            new NettyEncoder(),
                            new NettyDecoder(),
                            new IdleStateHandler(0, 0, nettyServerConfig.getServerChannelMaxIdleTimeSeconds()),
                            new NettyConnectManageHandler(),
                            new NettyServerHandler()
                        );
                }

//class HandshakeHandler extends SimpleChannelInboundHandler<ByteBuf>//TlsMode
//public class IdleStateHandler extends ChannelDuplexHandler
//连接管理：class NettyConnectManageHandler extends ChannelDuplexHandler
//真正处理业务：class NettyServerHandler extends SimpleChannelInboundHandler<RemotingCommand>

```

## 3. 设计特点

• - 远程业务都是内部使用，明确的类型与参数。所以有业务类型code与业务属性extFields两个部分，是一一对应的。
• - opaque:`是请求id，有异步请求必须，要等结果回来按id找请求。`
• - serializeTypeCurrentRPC:`因为给业务继续进行extFields的解析，所以设计的`
• - language与version：有语言与版本要求时使用
• - extFields：
•   `这个字段不通的请求/响应不一样，完全自定义。数据结构上是java的hashmap。在Java的每个RemotingCammand中，其实都带有一个CommandCustomHeader的属性成员，可以认为他是一个强类型的extFields，再最后传输的时候，这个CommandCustomHeader会被忽略，而传输前会把其中的所有字段全部都原封不动塞到extFields中，以作传输。`
• - remark:`附带的文本信息。常见的如存放一些broker/nameserver返回的一些异常信息，方便开发人员定位问题。`
• - flag： 按位(bit)解释。
•   `第0位标识是这次通信是request还是response，0标识request, 1 标识response。第1位标识是否是oneway请求，1标识oneway。应答方在处理oneway请求的时候，不会做出响应，请求方也无序等待应答方响应。`

# 三、我们项目的协议设计

## 1. 从dubbo与rocketmq看设计协议

两个协议的差别还是比较在的，再对比一下dubbo的字段：

```java
magic：类似java字节码文件里的魔数，用来判断是不是dubbo协议的数据包。魔数是常量0xdabb,用于判断报文的开始。
flag：标志位, 一共8个地址位。低四位用来表示消息体数据用的序列化工具的类型（默认hessian），高四位中，第一位为1表示是request请求，第二位为1表示双向传输（即有返回response），第三位为1表示是心跳ping事件。
status：状态位, 设置请求响应状态，dubbo定义了一些响应的类型。具体类型见
  com.alibaba.dubbo.remoting.exchange.Response
invoke id：消息id, long 类型。每一个请求的唯一识别id（由于采用异步通讯的方式，用来把请求request和返回的response对应上）
body length：消息体 body 长度, int 类型，即记录Body Content有多少个字节。
```

同时考虑一下http这样的最常见协议的设计，请求头的设计有这些考量：

• - id: 消息的id，异步必须使用。
• - 如果包含心跳，消息类型要区分心跳与业务。rocketmq业务种类是第二层的具体业务里的区分。
• - 请求还是响应，oneway还是twoway的标识。
• - 序列化类型，如果netty中完全解析了，就不用了。
• - magic，语言，版本适当考虑。
• - 响应时，对应的状态，比如http的200，500等一般放在头。
• - 消息长度。一般用netty，还是用它的基于长度的切分吧。头部是不是定长，要不要使用头部长度。body长度要不要？

## 2. 我们的api式消息协议

我们的要求是协议的充分安全，参考api接口的设计，我们每个客户端都有名有姓的，要进行登陆操作，还要有心跳操作，根据这些业务要求，通用的数据要求，在协议头中体现。

业务请求就是action，业务参数放在body中，服务端有一个action与业务serive对象的map，收到action调用service。

协议体就设计一个Object，我们把主要的信息都放头部，下面只介绍头。

另外对这些参数进行了类似api接口签名的操作。对body也用密钥进行了加密。

```java
//协议头的字段
public static final int VCODE= 0x202;//magic码
private int versionCode = VCODE;//版本号
private int length;//消息总长度
private String sessionID;//会话ID，一个登录成功的客户端，消息之间有sessionId确认。
private byte type;// 消息类型。见后面说明
private String appKey;// 客户端key
private String sign; //签名。MiddleMsg中的部分参数进行签名。类似于api请求中的签名。
//static String sign(String appSecret, MiddleMsg msg)
private long timestamp;//时间戳
private String action; //请求的业务操作
private String msgId ; //消息id
private int status; //消息状态
private Map<String, Object> attachment = new HashMap<String, Object>(); // 附加参数
```

```java
//netty处理的pipeline
new ChannelInitializer<SocketChannel>() {
    @Override
    public void initChannel(SocketChannel ch)
        throws IOException {
        ch.pipeline().addLast(new MessageDecoder(2000*1024 * 1024, 4, 4));//基于长度的。
        ch.pipeline().addLast(new MessageEncoder());
        //ch.pipeline().addLast("readTimeoutHandler",new ReadTimeoutHandler(120));
        ch.pipeline().addLast(new LoginAuthRespHandler());
        ch.pipeline().addLast("HeartBeatHandler",new HeartBeatRespHandler());
        ch.pipeline().addLast("handler", new ServiceHandler());
    }
```

```java
//type说明：
CLIENT_REQ((byte) 0),  //客户端请求 request
SERVICE_RESP((byte) 1), // response
ONE_WAY((byte) 2),  // 单线路
LOGIN_REQ((byte) 3), // 登录请求
LOGIN_RESP((byte) 4), // 登录响应
HEARTBEAT_REQ((byte) 5), // 心跳类型
HEARTBEAT_RESP((byte) 6), // 心跳响应
SERVICE_PUSH((byte) 7), //服务端推送
CLIENT_RESP_FOR_SERVICE_PUSH((byte) 8); //客户端的服务器回调
```

## 3. 密钥的交换

在LoginAuthRespHandler中，

一部分是基础校验：先检测IP，再检测用户名appKey。

上面通过了，如果是登录请求，再检测消息签名。正常登录成功后，产生一个sessionId，同时对应产生一个密钥管理类，初始为配置，里面同时产生下一步的密钥。下一次密钥作为返回消息体返回。

客户端定时心跳，每次心跳得到下一次要用的密钥，存起来。但当前的业务消息，始终是查当前的密钥用。





# 四、协议设计的思考

## 4.1 总体分析

远程传输的数据，有安全的要求，有请求/响应的区分，还有业务数据，通常用netty时，如何决定通讯协议呢？

rocketmq：在netty的decode/encode之前有一个HandshakeHandler，这属于在协议之外的处理，decode/encode之内的才是协议中的内容。消息的特点是更注重业务的处理，所以很多业务的数据都放在了协议头。再者终端用户的数据也可能复杂，所以不适合把rocketmq自己的业务数据放在body中混合在一起了。body纯放置终端用户的消息队列中的消息。netty消息handler中，只按head就把数据扔给业务了，不方便统一解析，每个业务自己解析body了。**head是基本的传输+内部业务调用参数map，body是终端用户数据。响应时就是常见的的code/msg+reslut（body中）**

dubbo：处理都在decode/encode之中，它的重点是传输rpcInvocation的信息，所以body中就是这个数据了，其它考虑放head中。由于支持配置的序列化协议，必须要在头中。响应码和响应文本按说可以放头部，这样有些时候不要看body数据了。但它放在了响应的body中，而rocketmq却有一个remark字段设计。前面说到可以类比http的api接口设计。**head是基本的传输，body是业务请求与结果。响应时外有服务result，body内有业务result。**

我们的协议：考虑对安全重视，除了业务数据外，要在头部放置安全处理要求的数据，就是把api的安全部分参数提升到了head中。还专门设计了netty中一个pipeline的handler处理安全部分，成功的才fire到后面处理。业务数据就都放在body中。只有action是对具体业务处理类的标识，放head中。**head是基本的传输+安全数据，body是业务请求或结果。**

## 4.2 结论

基本的传输都在head中：

• - 一般有请求/响应/心跳类型的标识，一般有oneway/twoday的标识，一般异步有id（一般AtomicInteger，不用uuid）的要求。
• - 至于magicCode/语言/协议版本/序列化看情况选择，序列化一般就定一种吧。
• - 另外总长度要有，用netty提供的基于长度的拆包用，头长度也有，拆包后要分head/body，分别解析出来。

核心业务看情况：

• - 有其它数据占用body的话只能放head中了。安全业务可以提升到head中，增加相关的字段。pipeline中也在真正业务处理器前，增加一个安全处理器。核心业务的数据或者安全方面可参考api接口的设计。

? 有点象应用中servlet中的filter，还有spring中的方法拦截器的设计一样，通用的东西要单独分出来。不过分了请求/响应/心跳类型，分了安全层，还有业务层，等于三个层，但协议只有head，body两层。把三层的东西放二层里，只能前两层的放head了，否则可以考虑增加一个neck(脖子)。