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

中国优秀的程序员网站程序员频道CXYCLUB技术地图
热搜:
更多>>
 
您所在的位置: 程序员俱乐部 > 编程开发 > JAVA > 基于netty的通讯协议的设计思考

基于netty的通讯协议的设计思考

 2019/12/29 15:40:07  herman_liu76  程序员俱乐部  我要评论(0)
  • 摘要:#一、dubbo协议的设计?先从业务开始,思考设计协议。##1.从rpc到remote?dubbo的目标是:一个接口方法的动态实现了远程方法调用,让调用者感觉与本地调用一样。既然是远程调用,涉及到用什么协议把调用数据发过去,以及接收方按这个协议解析出请求,进行处理后,用这个协议再返回结果。对于rpc要做的事情有:-调用方要做的事情有:产生一个包含调用数据的DubboInvoker,并且要用一个传输工具把这个发过去。-被调用方应该早就做好的事情:把自己所拥有的接口与实现
  • 标签:net 协议 设计
# 一、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(脖子)。
发表评论
用户名: 匿名