引言

本文主要从源码角度分析RocketMQ的底层通信机制以及RPC调用的过程。对于RocketMQ通信机制的深入理解，是我们分析和领会整个RocketMQ系统消息流转流程的基石。

消息
消息底层如何流转
总结

一、消息

我们可以设想一下，如果自己是RocketMQ的设计者，我们该如何设计一个消息系统。大家都知道RocketMQ是消息中间件，那么首先要解决的问题就是消息本身该如何设计。因为整个系统中它是传递的对象，是数据的载体。那么对于消息本身来说，我们需要定义消息的格式，这样客户端与服务端之间可以遵循定义好的消息格式来进行通信。在RocketMQ体系中，通过RemotingCommand对象来进行交互，，它对数据进行了封装。

（1）消息协议设计以及编解码

我们看下RemotingCommand中的部分属性如下所示：

private int code;   //操作码
private LanguageCode language = LanguageCode.JAVA;   //实现语言
private int version = 0;   //程序版本
private int opaque = requestId.getAndIncrement();   //reqeustId，在同一个连接上的不同请求标识码，与响应消息中的相对应
private int flag = 0;    //区分是普通RPC还是onewayRPC的标志
private String remark;    //传输自定义文本信息
private HashMap<String, String> extFields;    //自定义扩展信息
private transient CommandCustomHeader customHeader;

我们再看下具体的消息的格式是怎样的，如下图所示：

消息长度：总长度，四个字节存储，占用一个int类型；

序列化类型&消息头长度：同样占用一个int类型，第一个字节表示序列化类型，后面三个字节表示消息头长度；

消息头数据：经过序列化后的消息头数据；

消息主体数据：消息主体的二进制字节数据内容。

所以消息本身结构以及消息如何传递，都是我们需要了解的

图片来源于网络

我们看下remoting模块中的protocal文件夹下的相关定义类，如下所示：

对于消息还需要进行编码以及解码操作来提高消息传递的效率，所以RemotingCommand 中还提供了编码、解码的方法。RemotingCommand 的encode()源码如下所示：

 public ByteBuffer encode() {
        // 1> header length size
        int length = 4; //存储头部长度
        // 2> header data length
        byte[] headerData = this.headerEncode(); //报文头部数据
        length += headerData.length;
        // 3> body data length
        if (this.body != null) {
            length += body.length;
        }
        ByteBuffer result = ByteBuffer.allocate(4 + length);  //分配字节缓冲区
        // length
        result.putInt(length);
        // header length
        result.put(markProtocolType(headerData.length, serializeTypeCurrentRPC));
        // header data
        result.put(headerData);
        // body data;
        if (this.body != null) {
            result.put(this.body);
        }
        result.flip();  //将缓冲区翻转，用于将ByteBuffer放到网络通道中进行传输
        return result;
    }

我们再看下消息的解码操作，源码中的实现如下所示：

//解码操作
public static RemotingCommand decode(final byte[] array) {
        ByteBuffer byteBuffer = ByteBuffer.wrap(array);
        return decode(byteBuffer);
    }
//
public static RemotingCommand decode(final ByteBuffer byteBuffer) {
        int length = byteBuffer.limit();
        int oriHeaderLen = byteBuffer.getInt();
        int headerLength = getHeaderLength(oriHeaderLen);
        byte[] headerData = new byte[headerLength];
        byteBuffer.get(headerData);
        RemotingCommand cmd = headerDecode(headerData, getProtocolType(oriHeaderLen));
        int bodyLength = length - 4 - headerLength;
        byte[] bodyData = null;
        if (bodyLength > 0) {
            bodyData = new byte[bodyLength];
            byteBuffer.get(bodyData);
        }
        cmd.body = bodyData;
        return cmd;
    }

（2）消息传输方式

在RocketMQ系统中，消息的通信方式主要有三种：

1、可靠的同步传输；

2、可靠的异步传输；

3、单向传输；

二、消息如何流转

我们知道RocketMQ底层的通信是通过Netty实现。其中NettyRemotingClient与NettyRemotingServer是RemotingClient和RemotingServer这两个接口的实现类，也是比较重要的实现类。NettyRemotingClient与NettyRemotingServer也同时继承了NettyRemotingAbstract抽象类，该类抽象了invokeSync、invokeOneway等公用的方法实现。

（1）客户端请求发送

首先我们来看下 NettyRemotingClient中的重要源码分析，如下所示：

public class NettyRemotingClient extends NettyRemotingAbstract implements RemotingClient {
...
}

我们可以看出客户端继承了NettyRemotingAbstract抽象类同时实现了RemotingClient接口。其中比较重要的方法包括同步调用、异步调用以及单向调用。我么通过异步调用的方式来举例，NettyRemotingClient 根据地址信息获取或者创建channel，接着会进行invokeAsyncImpl方法的调用，此时客户端会将数据流转给公共的抽象类NettyRemotingAbstract 进行统一处理，在此类中完成真正的请求发送动作。发送消息的源码如下所示：

public void invokeAsyncImpl(final Channel channel, final RemotingCommand request, final long timeoutMillis,
        final InvokeCallback invokeCallback)
        throws InterruptedException, RemotingTooMuchRequestException, RemotingTimeoutException, RemotingSendRequestException {
        long beginStartTime = System.currentTimeMillis();
        //请求的ID
        final int opaque = request.getOpaque();
        boolean acquired = this.semaphoreAsync.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);
        if (acquired) {
            final SemaphoreReleaseOnlyOnce once = new SemaphoreReleaseOnlyOnce(this.semaphoreAsync);
            long costTime = System.currentTimeMillis() - beginStartTime;
            if (timeoutMillis < costTime) {
                once.release();
                throw new RemotingTimeoutException("invokeAsyncImpl call timeout");
            }
      //构建ResponseFuture 
            final ResponseFuture responseFuture = new ResponseFuture(channel, opaque, timeoutMillis - costTime, invokeCallback, once);
            //将ResponseFuture放入responseTable
            this.responseTable.put(opaque, responseFuture);
            try {
              //使用Netty的channel进行请求数据的发送
                channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
                  //消息发送后执行动作
                    @Override
                    public void operationComplete(ChannelFuture f) throws Exception {
                        if (f.isSuccess()) {
                          //如果消息发送成功，则直接将属性置为true，同时返回
                            responseFuture.setSendRequestOK(true);
                            return;
                        }
                        //如果失败
                        requestFail(opaque);
                        log.warn("send a request command to channel <{}> failed.", RemotingHelper.parseChannelRemoteAddr(channel));
                    }
                });
            } catch (Exception e) {
              //异常处理过程
                responseFuture.release();
                log.warn("send a request command to channel <" + RemotingHelper.parseChannelRemoteAddr(channel) + "> Exception", e);
                throw new RemotingSendRequestException(RemotingHelper.parseChannelRemoteAddr(channel), e);
            }
        } else {
            if (timeoutMillis <= 0) {
                throw new RemotingTooMuchRequestException("invokeAsyncImpl invoke too fast");
            } else {
                String info =
                    String.format("invokeAsyncImpl tryAcquire semaphore timeout, %dms, waiting thread nums: %d semaphoreAsyncValue: %d",
                        timeoutMillis,
                        this.semaphoreAsync.getQueueLength(),
                        this.semaphoreAsync.availablePermits()
                    );
                log.warn(info);
                throw new RemotingTimeoutException(info);
            }
        }
    }

此处有需要留意的部分，如下：

protected final ConcurrentMap<Integer /* opaque */, ResponseFuture> responseTable =
        new ConcurrentHashMap<Integer, ResponseFuture>(256);

opaque 表示请求码，用来标识不同的请求。这个请求码会与ResponseFuture映射在一起。

（2）服务端响应流程

服务端代码如下，继承了NettyRemotingAbstract同时实现了RemotingServer。

public class NettyRemotingServer extends NettyRemotingAbstract implements RemotingServer {
...
}

在该类中的start方法如下所示，它的主要作用是启动Netty服务器

@Override
    public void start() {
        this.defaultEventExecutorGroup = new DefaultEventExecutorGroup(
            nettyServerConfig.getServerWorkerThreads(),
            new ThreadFactory() {
                private AtomicInteger threadIndex = new AtomicInteger(0);
                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, "NettyServerCodecThread_" + this.threadIndex.incrementAndGet());
                }
            });
        ServerBootstrap childHandler =
            this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupSelector)
                .channel(useEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
                .option(ChannelOption.SO_BACKLOG, 1024)
                .option(ChannelOption.SO_REUSEADDR, true)
                .option(ChannelOption.SO_KEEPALIVE, false)
                .childOption(ChannelOption.TCP_NODELAY, true)
                .childOption(ChannelOption.SO_SNDBUF, nettyServerConfig.getServerSocketSndBufSize())
                .childOption(ChannelOption.SO_RCVBUF, nettyServerConfig.getServerSocketRcvBufSize())
                .localAddress(new InetSocketAddress(this.nettyServerConfig.getListenPort()))
                .childHandler(new ChannelInitializer<SocketChannel>() {
                  //向管道中添加数据处理逻辑
                    @Override
                    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()
                            );
                    }
                });
        if (nettyServerConfig.isServerPooledByteBufAllocatorEnable()) {
            childHandler.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
        }
        try {
            ChannelFuture sync = this.serverBootstrap.bind().sync();
            InetSocketAddress addr = (InetSocketAddress) sync.channel().localAddress();
            this.port = addr.getPort();
        } catch (InterruptedException e1) {
            throw new RuntimeException("this.serverBootstrap.bind().sync() InterruptedException", e1);
        }
        if (this.channelEventListener != null) {
            this.nettyEventExecutor.start();
        }
        this.timer.scheduleAtFixedRate(new TimerTask() {
            @Override
            public void run() {
                try {
                    NettyRemotingServer.this.scanResponseTable();
                } catch (Throwable e) {
                    log.error("scanResponseTable exception", e);
                }
            }
        }, 1000 * 3, 1000);
    }