RocketMq：04-网络通讯

1. 网络协议

客户端和服务端之间完成数据交互，需要约定数据协议。数据协议如下图所示：

传输内筒分为以下四个部分：

消息长度：
总长度，四个字节存储，占用一个int类型；
序列化类型&消息头长度：
占⽤⼀个 int 类型，第⼀个字节表示序列化类型，后⾯三个字节表示消息头⻓度；
消息头数据：
经过序列化后的消息头数据；
消息主体数据：
消息主体的⼆进制字节数据内容。

消息头数据序列化默认是 JSON 格式，示例如下：

⽹络协议设计的原则是便于编解码，这⾥我们温习下 TCP 粘包和拆包的知识点。

TCP 是⾯向字节流的协议，它会将应⽤层发送的数据拆分成 TCP 报⽂段进⾏传输，发送端和接收端都会维护⼀个 buffer ，发送的数据⾸先会存⾄缓冲区 buffer ，然后通过⽹络发送给接收端的 buffer 中。

粘包如果⼀次请求发送的数据量⽐较⼩，没达到缓冲区⼤⼩，TCP 则会将多个请求合并为同⼀个请求进⾏发送。
拆包如果⼀次请求发送的数据量⽐较⼤，超过了缓冲区⼤⼩，TCP 就会将其拆分为多次发送。

Netty 通过以下⼏种⽅式来解决粘包问题：

1、消息定⻓：FixedLengthFrameDecoder 发送的消息都是固定⻓度的，接收⽅根据固定⻓度来解析消息，这样可以有效避免粘包和拆包问题。

2、特定分隔符：DelimiterBasedFrameDecoder 在消息的末尾添加特定的分隔符，接收⽅根据分隔符来切分消息。

3、消息头⻓度：LenghtFieldBasedFrameDecode 在消息的头部添加表示消息⻓度的字段，接收⽅先读取消息头部的⻓度字段，然后根据⻓度字段的值来读取消息内容，从⽽正确地解析出完整的消息。

RocketMQ 的解码器就是使⽤了 LenghtFieldBasedFrameDecoder.

public class NettyDecoder extends LengthFieldBasedFrameDecoder {
    private static final InternalLogger log = InternalLoggerFactory.getLogger(RemotingHelper.ROCKETMQ_REMOTING);

    private static final int FRAME_MAX_LENGTH =
        Integer.parseInt(System.getProperty("com.rocketmq.remoting.frameMaxLength", "16777216"));

    public NettyDecoder() {
        super(FRAME_MAX_LENGTH, 0, 4, 0, 4);
    }

    @Override
    public Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
        ByteBuf frame = null;
        try {
            frame = (ByteBuf) super.decode(ctx, in);
            if (null == frame) {
                return null;
            }

            ByteBuffer byteBuffer = frame.nioBuffer();

            return RemotingCommand.decode(byteBuffer);
        } catch (Exception e) {
            log.error("decode exception, " + RemotingHelper.parseChannelRemoteAddr(ctx.channel()), e);
            RemotingUtil.closeChannel(ctx.channel());
        } finally {
            if (null != frame) {
                frame.release();
            }
        }

        return null;
    }
}

2. 通讯方式

客户端通信⽅式⽀持同步 sync 、异步 async 、单向 oneway 三种⽅式。

RemotingCommand invokeSync(
   	final String addr, 
   	final RemotingCommand request,
       final long timeoutMillis
) throws InterruptedException, RemotingConnectException,
       RemotingSendRequestException, RemotingTimeoutException;

   void invokeAsync(
       final String addr, 
       final RemotingCommand request, 
       final long timeoutMillis,
       final InvokeCallback invokeCallback
   ) throws InterruptedException, RemotingConnectException,
       RemotingTooMuchRequestException, RemotingTimeoutException, RemotingSendRequestException;

   void invokeOneway(
       final String addr, 
       final RemotingCommand request, 
       final long timeoutMillis
   ) throws InterruptedException, RemotingConnectException, RemotingTooMuchRequestException,
       RemotingTimeoutException, RemotingSendRequestException;

2.1 同步 sync

在同步通信中，客户端发送请求后会⼀直等待服务器响应，直到接收到响应或者超时。

这意味着：客户端发送线程在发送请求后会被阻塞，直到收到服务器的响应，然后继续执⾏发送下⼀个请求。

同步请求的流程：

1、客户端连接服务端，创建 channel ；

2、客户端创建 responseFutrue 对象，主要由四个部分组成：响应结果、请求编号、回调函数、CountDownLatch。然后将 responseFutrue 对象加⼊到本地缓存响应表 reponseTable ⾥。

1 2	final ResponseFuture responseFuture = new ResponseFuture(channel, opaque, timeoutMillis - costTime, invokeCallback, once); this.responseTable.put(opaque, responseFuture);

3、客户端将请求发送到服务端；

4、服务端解析出请求命令；

 1. 请求命令中包含命令类型、请求编号，服务端根据命令类型选择处理器 ，执⾏请求命令； 
 2.  服务端将响应数据返回给客户端； 
 3.  客户端将响应结果填充到响应表 reponseTable ⾥，同时因为是同步命令，并调⽤ countDownLatch 的 countDown ⽅法 , 这样发送消息线程就不再阻塞（实现同步请求的精髓）。

2.2 异步async

异步通信中，客户端发送请求后不会等待服务器的响应，⽽是继续执⾏后续代码。客户端会注册⼀个回调函数或者监听器，⽤于处理服务器响应。当服务器响应返回时，会触发回调函数的执⾏。

异步请求的流程：

1、客户端连接服务端，创建 channel ；

2、通过信号量 semaphoreAsync 限制正在进⾏的异步请求的最⼤数量 ; boolean

1	acquired = this.semaphoreAsync.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);

3、客户端创建 responseFutrue 对象，主要由四个部分组成：响应结果、请求编号、回调函数、CountDownLatch。然后将 responseFutrue 对象加⼊到本地缓存响应表 reponseTable ⾥。

1 2	final ResponseFuture responseFuture = new ResponseFuture(channel, opaque, timeoutMillis - costTime, invokeCallback, once); this.responseTable.put(opaque, responseFuture);