欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

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

程序员文章站 2022-03-04 23:17:28
...
# 一、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,它们就会用编码器转换/传输/再反转换:**

```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
```


基于netty的通讯协议的设计思考
            
    
    博客分类: 协议 netty协议java



```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有多少个字节。
```


基于netty的通讯协议的设计思考
            
    
    博客分类: 协议 netty协议java



## 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(脖子)。
相关标签: netty 协议 java