RocketMq中網絡通訊之服務端

 

• 一，Broker服務端入口（NettyServer端）

首先RocketMq網絡通訊採用的Netty通訊。服務端主要集中在Broker中。咱們先看一下Broker的啓動類BrokerStartup

顯然具體邏輯是在start方法裏面，下面是實現：

public void start() throws Exception { if (this.messageStore != null) { this.messageStore.start(); } if (this.remotingServer != null) { this.remotingServer.start(); } if (this.fastRemotingServer != null) { this.fastRemotingServer.start(); } if (this.fileWatchService != null) { this.fileWatchService.start(); } if (this.brokerOuterAPI != null) { this.brokerOuterAPI.start(); } if (this.pullRequestHoldService != null) { this.pullRequestHoldService.start(); } if (this.clientHousekeepingService != null) { this.clientHousekeepingService.start(); } if (this.filterServerManager != null) { this.filterServerManager.start(); } if (!messageStoreConfig.isEnableDLegerCommitLog()) { startProcessorByHa(messageStoreConfig.getBrokerRole()); handleSlaveSynchronize(messageStoreConfig.getBrokerRole()); } this.registerBrokerAll(true, false, true); this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { @Override public void run() { try { BrokerController.this.registerBrokerAll(true, false, brokerConfig.isForceRegister()); } catch (Throwable e) { log.error("registerBrokerAll Exception", e); } } }, 1000 * 10, Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)), TimeUnit.MILLISECONDS); if (this.brokerStatsManager != null) { this.brokerStatsManager.start(); } if (this.brokerFastFailure != null) { this.brokerFastFailure.start(); } }

能夠從名字大體猜出接收遠程消息是remotingServer.start()，點進去觀察一下其具體實現：

這裏看到咱們熟悉的面孔ServerBootStrap, 那麼能夠明確一點，咱們要知道的具體通訊協議實現，一定是寫在一個handler裏面的：

.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() ); }

從這些handler中，由名字能夠猜想，通訊消息的解析發生在NettyServerHandler，進入NettyServerHandler：

由上圖可知，它自己就是一個讀消息的Handler, 能夠看到的是接收的消息體是RemotingCommand。這個類必然就是整個RocketMq的通訊協議。

點進去看一下：

大體上看由code、Header、body以及一些metedata組成。其實全部的Rpc調用框架基本上都是這個設計思路。全部的請求必須繼承自某一個父類。

不過如今的微服務體系彷佛沒有這樣子作，多是出於不一樣的服務需求多樣性考慮，可是沒有統一的請求頭着實怪異,後續有時間看一下這方面的設計考慮。

至此，Rpc的Netty調用鏈基本結束。

•  二，RocketMq 通訊編碼 

由一的部分紅功定位到了接收消息的入口，本章着重講解其解析消息的細節實現。

rocketMq通訊協議Netty採用的是面向字節流的報文設計。在發送端，前4個字節存儲整個報文長度，緊接着4個字節存儲頭信息，而後緊接着發送body字節流。源碼以下：

public ByteBuffer encodeHeader(final int bodyLength) {
        // 定義頭4個字節儲存整個報文長度
        int length = 4;

        // 計算頭部長度
        byte[] headerData;
        headerData = this.headerEncode();
        length += headerData.length;

        // 計算body長度
        length += bodyLength;

　　　　　// 頭部信息：總體報文長度信息（4個字節） + 頭部數據（length-bodyLehth） 
        ByteBuffer result = ByteBuffer.allocate(4 + length - bodyLength);

        // 第一個4字節： 存放報文總體長度信息，從這裏咱們能夠看到meaasge的消息長度是有限制的
        result.putInt(length);

        // 第二個4字節： 第一個字節存放的是序列化類型，有Java或者RocketMq類型。後三個字節存放的是頭部數據長度
        result.put(markProtocolType(headerData.length, serializeTypeCurrentRPC));

        // 寫入頭部數據
        result.put(headerData);

        result.flip();

        return result;
    }

　　經過對編碼部分源碼學習，通常對字節的操做喜歡用位運算符，好比要整型的第三個字節，int >>>0xff & 0xff 便可。下面是rocketMq解析的部分示例代碼：

public static SerializeType getProtocolType(int source) {
        return SerializeType.valueOf((byte) ((source >> 24) & 0xFF));
    }

　　待續。。。。。。