rocketmq網(wǎng)絡(luò)部分的整體的架構(gòu)

remoting 模塊是 mq 的基礎(chǔ)通信模塊，理解通信層的原理對(duì)理解模塊間的交互很有幫助。RocketMQ Remoting 模塊底層基于 Netty 網(wǎng)絡(luò)庫(kù)驅(qū)動(dòng)，因此需要先了解一些基本的Netty原理。

Netty

Netty 使用 Reactor 模式，將監(jiān)聽(tīng)線(xiàn)程、IO 線(xiàn)程、業(yè)務(wù)邏輯線(xiàn)程隔離開(kāi)來(lái)。對(duì)每個(gè)連接，都對(duì)應(yīng)一個(gè) ChannelPipeline。ChannelPipeline 的默認(rèn)實(shí)現(xiàn) DefaultChannelPipeline 中用一個(gè)雙向鏈表儲(chǔ)存著若干 ChannelHandlerContext，每個(gè)ChannelHandlerContext 又對(duì)應(yīng)著一個(gè) ChannelHandler。鏈表的頭部是一個(gè) ChannelOutboundHandler：

class HeadContext extends AbstractChannelHandlerContext implements ChannelOutboundHandler {
  ...
}

尾部是一個(gè) ChannelInboundHandler：

class TailContext extends AbstractChannelHandlerContext implements ChannelInboundHandler {
          ...
}

這里的 Inbound 是指這個(gè) Handler 處理 外界觸發(fā) 的事件，典型的就是對(duì)端發(fā)送了數(shù)據(jù)過(guò)來(lái)； Outbound 是指事件是 自己觸發(fā) 的，比如向?qū)Χ税l(fā)送數(shù)據(jù)。同時(shí)，一個(gè) inbound 的事件將在 ChannelPipeline 的 ChannelHandlerContext 鏈表中從頭到尾傳播；而一個(gè) outbound 的事件將會(huì)在 ChannelPipeline 的 ChannelHandlerContext 鏈表中從尾向頭傳播。這樣，就能將數(shù)據(jù)解碼、數(shù)據(jù)處理、數(shù)據(jù)編碼等操作分散到不同的 ChannelHandler 中去了。

另外，RocketMQ 的協(xié)議格式如下，開(kāi)頭4字節(jié)表示整個(gè)消息長(zhǎng)度，隨后4字節(jié)表示頭部數(shù)據(jù)的長(zhǎng)度，最后就是消息體的長(zhǎng)度：

<4 byte length> <4 byte header length> <N byte header data> <N byte body data>

最后，我們?cè)賮?lái)看一下 RocketMQ remoting 部分的 UML 圖，了解一下其大概由哪些部分組成：

RocketMQ remoting

上圖這些類(lèi)中，最重要的是 NettyRemotingClient 和 NettyRemotingServer，它們的一些公共方法就被封裝在 NettyRemotingAbstract 中。

RemotingServer

有了上面的基本認(rèn)識(shí)，就可以開(kāi)始著手分析 RemotingServer 的源碼了。

啟動(dòng)

public void start() {
       ...
       ServerBootstrap childHandler =
                this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupSelector)
                        .channel(NioServerSocketChannel.class)
                        .option(ChannelOption.SO_BACKLOG, 1024)
                        .option(ChannelOption.SO_REUSEADDR, true)
                        .option(ChannelOption.SO_KEEPALIVE, false)
                        .childOption(ChannelOption.TCP_NODELAY, true)
                        .option(ChannelOption.SO_SNDBUF, nettyServerConfig.getServerSocketSndBufSize())
                        .option(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,
                                        new NettyEncoder(),
                                        new NettyDecoder(),
                                        new IdleStateHandler(0, 0, nettyServerConfig.getServerChannelMaxIdleTimeSeconds()),
                                        new NettyConnetManageHandler(),
                                        new NettyServerHandler());
                            }
                        });
  ...
        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);
        }
  ...
}

可以看到，在 NettyRemotingServer 的 start() 方法中，啟動(dòng)了 netty，使用成員變量 eventLoopGroupBoss 接受連接，使用 eventLoopGroupSelector 處理 IO，并且使用 defaultEventExecutorGroup 來(lái)處理 ChannelHandler 中的業(yè)務(wù)邏輯。nettyServerConfig 用來(lái)封裝對(duì) Netty 的配置信息，包括 SendBufSize、RcvBufSize 等。最重要的是，添加了 NettyEncoder、NettyDecoder、IdleStateHandler、NettyConnetManageHandler、NettyServerHandler 幾個(gè)ChannelHandler。

隨后，如果 channelEventListener 不為 null， 則啟動(dòng)一個(gè)專(zhuān)門(mén)的線(xiàn)程監(jiān)聽(tīng) Channel 的各種事件。

      if (this.channelEventListener != null) {
            this.nettyEventExecuter.start();
      }

這個(gè)類(lèi)主要是循環(huán)的從一個(gè) LinkedBlockingQueue 中讀取事件，而后調(diào)用 channelEventListener 的不同方法處理事件：

 class NettyEventExecuter extends ServiceThread {
        //使用一個(gè) LinkedBlockingQueue 來(lái)存儲(chǔ)待處理的 NettyEvent
        private final LinkedBlockingQueue<NettyEvent> eventQueue = new LinkedBlockingQueue<NettyEvent>();
        private final int maxSize = 10000;

        //添加待處理事件，如果隊(duì)列大小沒(méi)用超過(guò)限制，則將事件入隊(duì)
        public void putNettyEvent(final NettyEvent event) {
            if (this.eventQueue.size() <= maxSize) {
                this.eventQueue.add(event);
            } else {
                PLOG.warn("event queue size[{}] enough, so drop this event {}", this.eventQueue.size(), event.toString());
            }
        }

        @Override
        public void run() {
            PLOG.info(this.getServiceName() + " service started");

            final ChannelEventListener listener = NettyRemotingAbstract.this.getChannelEventListener();
            //循環(huán)讀取事件，并處理
            while (!this.isStopped()) {
                try {
                    NettyEvent event = this.eventQueue.poll(3000, TimeUnit.MILLISECONDS);
                    if (event != null && listener != null) {
                        switch (event.getType()) {
                            case IDLE:
                                listener.onChannelIdle(event.getRemoteAddr(), event.getChannel());
                                break;
                            case CLOSE:
                                listener.onChannelClose(event.getRemoteAddr(), event.getChannel());
                                break;
                            case CONNECT:
                                listener.onChannelConnect(event.getRemoteAddr(), event.getChannel());
                                break;
                            case EXCEPTION:
                                listener.onChannelException(event.getRemoteAddr(), event.getChannel());
                                break;
                            default:
                                break;
                        }
                    }
                } catch (Exception e) {
                    PLOG.warn(this.getServiceName() + " service has exception. ", e);
                }
            }

            PLOG.info(this.getServiceName() + " service end");
        }
  ...
}

隨后，則啟動(dòng)一個(gè)定時(shí)器，每隔一段時(shí)間查看 responseTable 是否有超時(shí)未回應(yīng)的請(qǐng)求，并完成一些清理工作，responseTable 的作用將在后文說(shuō)明發(fā)送請(qǐng)求過(guò)程時(shí)說(shuō)明：

       this.timer.scheduleAtFixedRate(new TimerTask() {

            @Override
            public void run() {
                try {
                    NettyRemotingServer.this.scanResponseTable();
                } catch (Exception e) {
                    log.error("scanResponseTable exception", e);
                }
            }
        }, 1000 * 3, 1000);

至此，NettyRemotingServer 的啟動(dòng)過(guò)程就結(jié)束了。

ChannelHandler

在啟動(dòng)時(shí)，向 ChannelPipeline 中添加了以下ChannelHandler，我們分別來(lái)解釋其作用。

NettyEncoder

對(duì)發(fā)送請(qǐng)求按照上文提到的格式進(jìn)行編碼，沒(méi)用什么特殊的：

    @Override
    public void encode(ChannelHandlerContext ctx, RemotingCommand remotingCommand, ByteBuf out)
        throws Exception {
        try {
            ByteBuffer header = remotingCommand.encodeHeader();
            out.writeBytes(header);
            byte[] body = remotingCommand.getBody();
            if (body != null) {
                out.writeBytes(body);
            }
        } catch (Exception e) {
            log.error("encode exception, " + RemotingHelper.parseChannelRemoteAddr(ctx.channel()), e);
            if (remotingCommand != null) {
                log.error(remotingCommand.toString());
            }
            RemotingUtil.closeChannel(ctx.channel());
        }
    }

NettyDecoder

與 NettyEncoder 相反，這是一個(gè) Inbound ChannelHandler，對(duì)接收到的數(shù)據(jù)進(jìn)行解碼，注意由于 RocketMQ 的協(xié)議的頭部是定長(zhǎng)的，所以它繼承了 LengthFieldBasedFrameDecoder：

 @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;
    }

IdleStateHandler

這個(gè) Handler 是用來(lái)進(jìn)行 keepalive 的，當(dāng)一段時(shí)間沒(méi)有發(fā)送或接收到數(shù)據(jù)時(shí)，則觸發(fā) IdleStateEvent。

  protected void channelIdle(ChannelHandlerContext ctx, IdleStateEvent evt) throws Exception {
        ctx.fireUserEventTriggered(evt);
  }

NettyConnetManageHandler

負(fù)責(zé)處理各種連接事件，尤其是 IdleState，將其交給 channelEventListener 處理。

IdleStateEvent evnet = (IdleStateEvent) evt;
if ( evnet.state().equals( IdleState.ALL_IDLE ) )
{
    final String remoteAddress = RemotingHelper.parseChannelRemoteAddr( ctx.channel() );
    log.warn( "NETTY SERVER PIPELINE: IDLE exception [{}]", remoteAddress );
    RemotingUtil.closeChannel( ctx.channel() );
    if ( NettyRemotingServer.this.channelEventListener != null )
    {
        NettyRemotingServer.this
        .putNettyEvent( new NettyEvent( NettyEventType.IDLE, remoteAddress.toString(), ctx.channel() ) );
    }
}

NettyServerHandler

調(diào)用 NettyRemotingAbstract 的 processMessageReceived 方法處理請(qǐng)求。

class NettyServerHandler extends SimpleChannelInboundHandler<RemotingCommand> {
    @Override
    protected void channelRead0( ChannelHandlerContext ctx, RemotingCommand msg ) throws Exception
    {
        processMessageReceived( ctx, msg );
    }
}

public void processMessageReceived( ChannelHandlerContext ctx, RemotingCommand msg ) throws Exception
{
    final RemotingCommand cmd = msg;
    if ( cmd != null )
    {
        switch ( cmd.getType() )
        {
        case REQUEST_COMMAND:
            processRequestCommand( ctx, cmd );
            break;
        case RESPONSE_COMMAND:
            processResponseCommand( ctx, cmd );
            break;
        default:
            break;
        }
    }
}

在這里，請(qǐng)求可以分為兩類(lèi)，一類(lèi)是處理別的服務(wù)發(fā)來(lái)的請(qǐng)求；另外一類(lèi)是處理自己發(fā)給別的服務(wù)的請(qǐng)求的處理結(jié)果。所有的請(qǐng)求其實(shí)都是異步的，只是將請(qǐng)求相關(guān)的 ResponseFuture記在一個(gè) ConcurrentHashMap 中，map 的 key 為與請(qǐng)求相關(guān)的一個(gè)整數(shù)。

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

另外需要注意的時(shí)，對(duì)不同類(lèi)型的請(qǐng)求（由 RemotingCommand 的 code 字段標(biāo)識(shí)），會(huì)提前注冊(cè)對(duì)應(yīng)的 NettyRequestProcessor 以及 ExecutorService，對(duì)請(qǐng)求的處理將放在注冊(cè)好的線(xiàn)程池中進(jìn)行：

 public void processRequestCommand(final ChannelHandlerContext ctx, final RemotingCommand cmd) {
        final Pair<NettyRequestProcessor, ExecutorService> matched = this.processorTable.get(cmd.getCode());
        final Pair<NettyRequestProcessor, ExecutorService> pair = null == matched ? this.defaultRequestProcessor : matched;
        final int opaque = cmd.getOpaque();
        ...
}

對(duì)于 ResponseRequest 的處理則較為簡(jiǎn)單，只是將其從 responseTable 中刪掉，然后再調(diào)用 ResponseFuture 的 putResponse 方法設(shè)置返回結(jié)果，或是調(diào)用 responseFuture 中預(yù)設(shè)的回掉方法。

  public void processResponseCommand(ChannelHandlerContext ctx, RemotingCommand cmd) {
        final int opaque = cmd.getOpaque();
        final ResponseFuture responseFuture = responseTable.get(opaque);
        if (responseFuture != null) {
            responseFuture.setResponseCommand(cmd);

            responseFuture.release();

            responseTable.remove(opaque);

            if (responseFuture.getInvokeCallback() != null) {
                executeInvokeCallback(responseFuture);
            } else {
                responseFuture.putResponse(cmd);
            }
        } else {
            PLOG.warn("receive response, but not matched any request, " + RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
            PLOG.warn(cmd.toString());
        }
    }

向其他服務(wù)發(fā)起請(qǐng)求

請(qǐng)求有三種，分別是異步請(qǐng)求、同步請(qǐng)求以及單向請(qǐng)求，分別調(diào)用了 NettyRemotingAbstract 的對(duì)應(yīng)方法。從上文的分析我們可以看到，異步請(qǐng)求其實(shí)是使用 opaque 字段標(biāo)識(shí)了一次請(qǐng)求，然后生成一個(gè)占位符 ResponseFuture 并存儲(chǔ)起來(lái)。接收方在處理完請(qǐng)求后，發(fā)送一個(gè)相同 opaque 值的回應(yīng)請(qǐng)求，從而通過(guò) opaque 找到對(duì)應(yīng)的 ResponseFuture，返回結(jié)果或是運(yùn)行預(yù)設(shè)的回調(diào)函數(shù)。同步請(qǐng)求其實(shí)也是一個(gè)異步請(qǐng)求，只不過(guò)通過(guò) CountdownLatch 使調(diào)用者發(fā)生阻塞。單向請(qǐng)求最簡(jiǎn)單，只發(fā)送，不關(guān)注請(qǐng)求結(jié)果。

下面以 invokeAsync 為例分析整個(gè)過(guò)程：

//調(diào)用 NettyRemotingAbstract 的 invokeAsyncImpl 方法
@Override
public void invokeAsync(Channel channel, RemotingCommand request, long timeoutMillis, InvokeCallback invokeCallback)
            throws InterruptedException, RemotingTooMuchRequestException, RemotingTimeoutException, RemotingSendRequestException {
        this.invokeAsyncImpl(channel, request, timeoutMillis, invokeCallback);
}

public void invokeAsyncImpl( final Channel channel, final RemotingCommand request, final long timeoutMillis,
             final InvokeCallback invokeCallback )
throws InterruptedException, RemotingTooMuchRequestException, RemotingTimeoutException, RemotingSendRequestException
{
    final int   opaque      = request.getOpaque();
    boolean     acquired    = this.semaphoreAsync.tryAcquire( timeoutMillis, TimeUnit.MILLISECONDS );
    if ( acquired )
    {
        final SemaphoreReleaseOnlyOnce once = new SemaphoreReleaseOnlyOnce( this.semaphoreAsync );

        final ResponseFuture responseFuture = new ResponseFuture( opaque, timeoutMillis, invokeCallback, once );
        this.responseTable.put( opaque, responseFuture );
        try {
            channel.writeAndFlush( request ).addListener( new ChannelFutureListener()
                                 {
                                     @Override
                                     public void operationComplete( ChannelFuture f ) throws Exception {
                                         if ( f.isSuccess() )
                                         {
                                             responseFuture.setSendRequestOK( true );
                                             return;
                                         } else {
                                             responseFuture.setSendRequestOK( false );
                                         }

                                         responseFuture.putResponse( null );
                                         responseTable.remove( opaque );
                                         try {
                                             executeInvokeCallback( responseFuture );
                                         } catch ( Throwable e ) {
                                             PLOG.warn( "excute callback in writeAndFlush addListener, and callback throw", e );
                                         } finally {
                                             responseFuture.release();
                                         }

                                         PLOG.warn( "send a request command to channel <{}> failed.", RemotingHelper.parseChannelRemoteAddr( channel ) );
                                          }
                                      } );
        } catch ( Exception e ) {
            responseFuture.release();
            PLOG.warn( "send a request command to channel <" + RemotingHelper.parseChannelRemoteAddr( channel ) + "> Exception", e );
            throw new RemotingSendRequestException( RemotingHelper.parseChannelRemoteAddr( channel ), e );
        }
    } else {
        String info =
            String.format( "invokeAsyncImpl tryAcquire semaphore timeout, %dms, waiting thread nums: %d semaphoreAsyncValue: %d",   /*  */
                       timeoutMillis,                                                                                           /*  */
                       this.semaphoreAsync.getQueueLength(),                                                                    /*  */
                       this.semaphoreAsync.availablePermits()                                                                   /*  */
                       );
        PLOG.warn( info );
        throw new RemotingTooMuchRequestException( info );
    }
}

在請(qǐng)求時(shí)如果失敗成功則直接返回，如果失敗則從 responseTable 刪除本次請(qǐng)求，并調(diào)用 responseFuture.putResponse( null )，然后執(zhí)行失敗回調(diào) executeInvokeCallback( responseFuture )。而后，就是等待對(duì)方發(fā)來(lái)的 Response Request 了，上文已經(jīng)有過(guò)分析，這里不再贅述。

下面，看看同步消息。在發(fā)送請(qǐng)求后，調(diào)用了 ResponseFuture 的 waitResponse 方法。這個(gè)方法調(diào)用了 CountDownLatch 的 await 方法。請(qǐng)求處理成功或失敗后則會(huì)調(diào)用 ResponseFuture 的 putResponse 方法，設(shè)置處理結(jié)果并打開(kāi) CountDownLatch，從而實(shí)現(xiàn)了同步調(diào)用。

   public RemotingCommand invokeSyncImpl(final Channel channel, final RemotingCommand request, final long timeoutMillis)
        throws InterruptedException, RemotingSendRequestException, RemotingTimeoutException {
        final int opaque = request.getOpaque();

        try {
            final ResponseFuture responseFuture = new ResponseFuture(opaque, timeoutMillis, null, null);
            this.responseTable.put(opaque, responseFuture);
            final SocketAddress addr = channel.remoteAddress();
            channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture f) throws Exception {
                    if (f.isSuccess()) {
                        responseFuture.setSendRequestOK(true);
                        return;
                    } else {
                        responseFuture.setSendRequestOK(false);
                    }

                    responseTable.remove(opaque);
                    responseFuture.setCause(f.cause());
                    responseFuture.putResponse(null);
                    PLOG.warn("send a request command to channel <" + addr + "> failed.");
                }
            });

            RemotingCommand responseCommand = responseFuture.waitResponse(timeoutMillis);
            if (null == responseCommand) {
                if (responseFuture.isSendRequestOK()) {
                    throw new RemotingTimeoutException(RemotingHelper.parseSocketAddressAddr(addr), timeoutMillis,
                        responseFuture.getCause());
                } else {
                    throw new RemotingSendRequestException(RemotingHelper.parseSocketAddressAddr(addr), responseFuture.getCause());
                }
            }

            return responseCommand;
        } finally {
            this.responseTable.remove(opaque);
        }
    }

   public RemotingCommand waitResponse(final long timeoutMillis) throws InterruptedException {
        this.countDownLatch.await(timeoutMillis, TimeUnit.MILLISECONDS);
        return this.responseCommand;
   }

   public void putResponse(final RemotingCommand responseCommand) {
        this.responseCommand = responseCommand;
        this.countDownLatch.countDown();
   }

NettyRemotingClient 的思路與 NettyRemotingServer 類(lèi)似，這里不再進(jìn)行分析。

以上。