RocketMq通信基础知识

1 通信组成

Rocketmq的通信层是基于通信框架netty 4.0.21.Final之上做了简单的协议封装。不过在目前的版本中(v3.2.6)，netty不再是强耦合与通信模块，基本的类图如下： popo_2018-09-23 20-23-37.jpg Producer与Namesrv：Producer作为client，Namesrv作为server，例如，GET_ROUTEINTO_BY_TOPIC； Consumer与Namesrv：Consumer作为client，Namesrv作为server； Broker与Namesrv：Broker作为client，Namesrv作为server，例如，REGISTER_BROKER，UNREGISTER_BROKER； Producer与Broker：Producer作为client，Broker作为server，例如，CONSUMER_SEND_MSG_BACK； Consumer与Broker：Consumer作为client，Broker作为server。 主要通信模块包括Namesrv、Broker、Producer和Consumer模块。

2 消息格式

各个模块处理的消息请求和响应格式都是RemotingCommand，消息头根据不同的消息码抽象为CommandCustomHeader接口的实现类。 由于消息体内部使用4字节表示消息序列化方式(JSON/ROCKETMQ)和消息总长度，所以RocketMQ支持的消息最大长度为2的24次方，即16777216。int的最高8位会丢弃。消息头内的数据为RemotingCommand对象的属性以及一个Map。

通过org.apache.rocketmq.remoting.netty.NettyEncoder和org.apache.rocketmq.remoting.netty.NettyDecoder对消息进行加解密。

消息格式为(数字为几个字节)：

总长度为：4+4+头数据长度+消息体数据长度，默认情况下，消息的最大长度为：16777216，即2的24次方。 序列化类型：0为JSON，1为ROCKETMQ。JSON格式可以很好的跨语言。 头长度为头数据的长度。 头数据保存RemotingCommand对象属性以及一个Map(extFields)，extFields用于保存CommandCustomHeader子类的属性。消息头和消息码一般是一一对应的，且为硬编码 。 消息体用于保存消息业务数据，使用字节传输。 消息头数据header data

3 各通信模块介绍

3.1 Namesrv模块

Namersrv主要作为netty的服务端server，进行请求的处理。请求是通过netty的Handler注册到Netty中，处理类为NettyRequestProcessor接口的实现类，NameServer的默认处理类为：DefaultRequestProcessor。Namesrv的初始化过程主要在 org.apache.rocketmq.namesrv.NamesrvController#initialize 中完成：

    private void registerProcessor() {
        if (namesrvConfig.isClusterTest()) {

            this.remotingServer.registerDefaultProcessor(new ClusterTestRequestProcessor(this, namesrvConfig.getProductEnvName()),
                this.remotingExecutor);
        } else {

            this.remotingServer.registerDefaultProcessor(new DefaultRequestProcessor(this), this.remotingExecutor);
        }
    }

    public void start() throws Exception {
        this.remotingServer.start();

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

3.2 Broker模块

3.2.1 netty client

Broker作为netty客户端，主要用于从Namesrv注册一个broker或者卸载一个broker。 初始化client

    public BrokerOuterAPI(final NettyClientConfig nettyClientConfig, RPCHook rpcHook) {
        this.remotingClient = new NettyRemotingClient(nettyClientConfig);
        this.remotingClient.registerRPCHook(rpcHook);
    }

发送向Namesrv注册broker请求

private RegisterBrokerResult registerBroker(
        final String namesrvAddr,
        final boolean oneway,
        final int timeoutMills,
        final RegisterBrokerRequestHeader requestHeader,
        final byte[] body
    ) throws RemotingCommandException, MQBrokerException, RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException,
        InterruptedException {
        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.REGISTER_BROKER, requestHeader);
        request.setBody(body);

        if (oneway) {
            try {
                this.remotingClient.invokeOneway(namesrvAddr, request, timeoutMills);
            } catch (RemotingTooMuchRequestException e) {
                // Ignore
            }
            return null;
        }

        RemotingCommand response = this.remotingClient.invokeSync(namesrvAddr, request, timeoutMills);
        assert response != null;
        switch (response.getCode()) {
            case ResponseCode.SUCCESS: {
                RegisterBrokerResponseHeader responseHeader =
                    (RegisterBrokerResponseHeader) response.decodeCommandCustomHeader(RegisterBrokerResponseHeader.class);
                RegisterBrokerResult result = new RegisterBrokerResult();
                result.setMasterAddr(responseHeader.getMasterAddr());
                result.setHaServerAddr(responseHeader.getHaServerAddr());
                if (response.getBody() != null) {
                    result.setKvTable(KVTable.decode(response.getBody(), KVTable.class));
                }
                return result;
            }
            default:
                break;
        }

        throw new MQBrokerException(response.getCode(), response.getRemark());
    }

3.2.2 netty server

Broker作为netty服务端，主要用于处理消息生产者和消费者的请求。

3.2.2.1 处理Producer消息

Broker通过注册SendMessageProcessor作为收到生产者消息后的处理实现类，注册过程如下：

public void registerProcessor() {
        /**
         * SendMessageProcessor
         */
        SendMessageProcessor sendProcessor = new SendMessageProcessor(this);
        sendProcessor.registerSendMessageHook(sendMessageHookList);
        sendProcessor.registerConsumeMessageHook(consumeMessageHookList);

        this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE, sendProcessor, this.sendMessageExecutor);
        this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE_V2, sendProcessor, this.sendMessageExecutor);
        this.remotingServer.registerProcessor(RequestCode.SEND_BATCH_MESSAGE, sendProcessor, this.sendMessageExecutor);
        this.remotingServer.registerProcessor(RequestCode.CONSUMER_SEND_MSG_BACK, sendProcessor, this.sendMessageExecutor);
        this.fastRemotingServer.registerProcessor(RequestCode.SEND_MESSAGE, sendProcessor, this.sendMessageExecutor);
        this.fastRemotingServer.registerProcessor(RequestCode.SEND_MESSAGE_V2, sendProcessor, this.sendMessageExecutor);
        this.fastRemotingServer.registerProcessor(RequestCode.SEND_BATCH_MESSAGE, sendProcessor, this.sendMessageExecutor);
        this.fastRemotingServer.registerProcessor(RequestCode.CONSUMER_SEND_MSG_BACK, sendProcessor, this.sendMessageExecutor);
        ...
}

broker主要通过SendMessageProcessor.processRequest()处理Producer生成的消息。该类是负责响应Producer发消息到broker的入口处理逻辑类。其processRequest响应两种请求：

• CONSUMER_SEND_MSG_BACK： consumer消费失败的消息发回broker。
• default： producer发消息到broker。

下章重点讲解broker消息处理过程。

3.2.2.2 处理Consumer消息

消息消费处理主要由PullMessageProcessor.processRequest处理。

public void registerProcessor() {
        ...
        /**
         * PullMessageProcessor
         */
        this.remotingServer.registerProcessor(RequestCode.PULL_MESSAGE, this.pullMessageProcessor, this.pullMessageExecutor);
        this.pullMessageProcessor.registerConsumeMessageHook(consumeMessageHookList);

        ...
}

3.3 Producer

Producer主要作为netty客户端，用于与Namesrv和Broker进行通信。和Namesrv通信主要用于获取Broker的信息；和Broker通信主要用于消息的发送。 在需要发送消息时，会启动netty客户端，例如：

DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
        producer.start();

3.4 Consumer

Consumer作为netty客户端，用于与Namesrv和Broker进行通信。和Namesrv通信主要用于获取Broker的信息；和Broker通信主要用于获取消息来消费。

4 通信模板启动

4.1 服务端的启动操作

remoting模块一般是由initialize()和start()两部分组成 initialize部分做的事情:

• 初始化netty引导(serverBootStrap)，初始化boss线程池、worker线程池。
• 初始化ChannelEvent监听(这个用于监听channelEvent事件的到达)和ChannelEvent执行器
• 注册通讯命令(RemotingCommand)解析器，这个解析器用于解释服务端接受到的命令(rocketMQ中的通讯消息全部由通讯命令封装，不同的通讯命令由不同的通讯命令类型构成)。

start部分做的事情：

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

• 启动netty引导程序(bootstrap)，设置事件循环组(EventLoopGroup)boss线程池，worker线程池。
• 启动ChannelEvent执行器

4.2 客户端的启动操作

同服务端模块，客户端模块也是由initialize()和start()组成 initialize部分做的事情：

• 初始化netty引导(serverBootStrap),初始化worker线程池(netty的客户端只有工作线程)
• 初始化ChannelEvent监听(这个用于监听channelEvent事件的到达)

start部分做的事情：

public void start() {
        this.defaultEventExecutorGroup = new DefaultEventExecutorGroup(
            nettyClientConfig.getClientWorkerThreads(),
            new ThreadFactory() {

                private AtomicInteger threadIndex = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, "NettyClientWorkerThread_" + this.threadIndex.incrementAndGet());
                }
            });

        Bootstrap handler = this.bootstrap.group(this.eventLoopGroupWorker).channel(NioSocketChannel.class)
            .option(ChannelOption.TCP_NODELAY, true)
            .option(ChannelOption.SO_KEEPALIVE, false)
            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, nettyClientConfig.getConnectTimeoutMillis())
            .option(ChannelOption.SO_SNDBUF, nettyClientConfig.getClientSocketSndBufSize())
            .option(ChannelOption.SO_RCVBUF, nettyClientConfig.getClientSocketRcvBufSize())
            .handler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline pipeline = ch.pipeline();
                    if (nettyClientConfig.isUseTLS()) {
                        if (null != sslContext) {
                            pipeline.addFirst(defaultEventExecutorGroup, "sslHandler", sslContext.newHandler(ch.alloc()));
                            log.info("Prepend SSL handler");
                        } else {
                            log.warn("Connections are insecure as SSLContext is null!");
                        }
                    }
                    pipeline.addLast(
                        defaultEventExecutorGroup,
                        new NettyEncoder(),
                        new NettyDecoder(),
                        new IdleStateHandler(0, 0, nettyClientConfig.getClientChannelMaxIdleTimeSeconds()),
                        new NettyConnectManageHandler(),
                        new NettyClientHandler());
                }
            });

        this.timer.scheduleAtFixedRate(new TimerTask() {
            @Override
            public void run() {
                try {
                    NettyRemotingClient.this.scanResponseTable();
                } catch (Throwable e) {
                    log.error("scanResponseTable exception", e);
                }
            }
        }, 1000 * 3, 1000);

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

• 初始化了默认事件执行器线程池(defaultEventExecutorGroup)
• 启动netty引导程序(bootstrap),设置事件循环组(EventLoopGroup)worker线程池
• 打开一个定时调度线程，定时查看超时的缓存请求，有callback的执行callback，然后从缓存中移除再释放请求

我们理解您需要更便捷更高效的工具记录思想，整理笔记、知识，并将其中承载的价值传播给他人，Cmd Markdown 是我们给出的答案 —— 我们为记录思想和分享知识提供更专业的工具。 您可以使用 Cmd Markdown：

• 整理知识，学习笔记
• 发布日记，杂文，所见所想
• 撰写发布技术文稿（代码支持）
• 撰写发布学术论文（LaTeX 公式支持）

除了您现在看到的这个 Cmd Markdown 在线版本，您还可以前往以下网址下载：

Windows/Mac/Linux 全平台客户端

请保留此份 Cmd Markdown 的欢迎稿兼使用说明，如需撰写新稿件，点击顶部工具栏右侧的 新文稿 或者使用快捷键 Ctrl+Alt+N。

什么是 Markdown

Markdown 是一种方便记忆、书写的纯文本标记语言，用户可以使用这些标记符号以最小的输入代价生成极富表现力的文档：譬如您正在阅读的这份文档。它使用简单的符号标记不同的标题，分割不同的段落，粗体 或者 斜体 某些文字，更棒的是，它还可以

1. 制作一份待办事宜 Todo 列表

• [ ] 支持以 PDF 格式导出文稿
• [ ] 改进 Cmd 渲染算法，使用局部渲染技术提高渲染效率
• [x] 新增 Todo 列表功能
• [x] 修复 LaTex 公式渲染问题
• [x] 新增 LaTex 公式编号功能

2. 书写一个质能守恒公式^LaTeX

$$E=mc^2$$

3. 高亮一段代码^code

@requires_authorization
class SomeClass:
    pass

if __name__ == '__main__':
    # A comment
    print 'hello world'

4. 高效绘制 流程图

st=>start: Start
op=>operation: Your Operation
cond=>condition: Yes or No?
e=>end

st->op->cond
cond(yes)->e
cond(no)->op

5. 高效绘制 序列图

Alice->Bob: Hello Bob, how are you?
Note right of Bob: Bob thinks
Bob-->Alice: I am good thanks!

6. 高效绘制 甘特图

    title 项目开发流程
    section 项目确定
        需求分析       :a1, 2016-06-22, 3d
        可行性报告     :after a1, 5d
        概念验证       : 5d
    section 项目实施
        概要设计      :2016-07-05  , 5d
        详细设计      :2016-07-08, 10d
        编码          :2016-07-15, 10d
        测试          :2016-07-22, 5d
    section 发布验收
        发布: 2d
        验收: 3d

7. 绘制表格

8. 更详细语法说明

想要查看更详细的语法说明，可以参考我们准备的 Cmd Markdown 简明语法手册，进阶用户可以参考 Cmd Markdown 高阶语法手册 了解更多高级功能。

总而言之，不同于其它 所见即所得 的编辑器：你只需使用键盘专注于书写文本内容，就可以生成印刷级的排版格式，省却在键盘和工具栏之间来回切换，调整内容和格式的麻烦。Markdown 在流畅的书写和印刷级的阅读体验之间找到了平衡。 目前它已经成为世界上最大的技术分享网站 GitHub 和 技术问答网站 StackOverFlow 的御用书写格式。

什么是 Cmd Markdown

您可以使用很多工具书写 Markdown，但是 Cmd Markdown 是这个星球上我们已知的、最好的 Markdown 工具——没有之一 ：）因为深信文字的力量，所以我们和你一样，对流畅书写，分享思想和知识，以及阅读体验有极致的追求，我们把对于这些诉求的回应整合在 Cmd Markdown，并且一次，两次，三次，乃至无数次地提升这个工具的体验，最终将它演化成一个 编辑/发布/阅读 Markdown 的在线平台——您可以在任何地方，任何系统/设备上管理这里的文字。

1. 实时同步预览

我们将 Cmd Markdown 的主界面一分为二，左边为编辑区，右边为预览区，在编辑区的操作会实时地渲染到预览区方便查看最终的版面效果，并且如果你在其中一个区拖动滚动条，我们有一个巧妙的算法把另一个区的滚动条同步到等价的位置，超酷！

2. 编辑工具栏

也许您还是一个 Markdown 语法的新手，在您完全熟悉它之前，我们在 编辑区 的顶部放置了一个如下图所示的工具栏，您可以使用鼠标在工具栏上调整格式，不过我们仍旧鼓励你使用键盘标记格式，提高书写的流畅度。

3. 编辑模式

完全心无旁骛的方式编辑文字：点击 编辑工具栏 最右侧的拉伸按钮或者按下 Ctrl + M，将 Cmd Markdown 切换到独立的编辑模式，这是一个极度简洁的写作环境，所有可能会引起分心的元素都已经被挪除，超清爽！

4. 实时的云端文稿

为了保障数据安全，Cmd Markdown 会将您每一次击键的内容保存至云端，同时在 编辑工具栏 的最右侧提示 已保存 的字样。无需担心浏览器崩溃，机器掉电或者地震，海啸——在编辑的过程中随时关闭浏览器或者机器，下一次回到 Cmd Markdown 的时候继续写作。

5. 离线模式

在网络环境不稳定的情况下记录文字一样很安全！在您写作的时候，如果电脑突然失去网络连接，Cmd Markdown 会智能切换至离线模式，将您后续键入的文字保存在本地，直到网络恢复再将他们传送至云端，即使在网络恢复前关闭浏览器或者电脑，一样没有问题，等到下次开启 Cmd Markdown 的时候，她会提醒您将离线保存的文字传送至云端。简而言之，我们尽最大的努力保障您文字的安全。

6. 管理工具栏

为了便于管理您的文稿，在 预览区 的顶部放置了如下所示的 管理工具栏：

通过管理工具栏可以：

</i> 发布：将当前的文稿生成固定链接，在网络上发布，分享 新建：开始撰写一篇新的文稿 </i> 删除：删除当前的文稿 导出：将当前的文稿转化为 Markdown 文本或者 Html 格式，并导出到本地 </i> 列表：所有新增和过往的文稿都可以在这里查看、操作 模式：切换 普通/Vim/Emacs 编辑模式

7. 阅读工具栏

通过 预览区 右上角的 阅读工具栏，可以查看当前文稿的目录并增强阅读体验。

工具栏上的五个图标依次为：

</i> 目录：快速导航当前文稿的目录结构以跳转到感兴趣的段落 视图：互换左边编辑区和右边预览区的位置 </i> 主题：内置了黑白两种模式的主题，试试 黑色主题，超炫！ 阅读：心无旁骛的阅读模式提供超一流的阅读体验 全屏：简洁，简洁，再简洁，一个完全沉浸式的写作和阅读环境

8. 阅读模式

在 阅读工具栏 点击 或者按下 Ctrl+Alt+M 随即进入独立的阅读模式界面，我们在版面渲染上的每一个细节：字体，字号，行间距，前背景色都倾注了大量的时间，努力提升阅读的体验和品质。

9. 标签、分类和搜索

在编辑区任意行首位置输入以下格式的文字可以标签当前文档：

标签： 未分类

标签以后的文稿在【文件列表】（Ctrl+Alt+F）里会按照标签分类，用户可以同时使用键盘或者鼠标浏览查看，或者在【文件列表】的搜索文本框内搜索标题关键字过滤文稿，如下图所示：

10. 文稿发布和分享

在您使用 Cmd Markdown 记录，创作，整理，阅读文稿的同时，我们不仅希望它是一个有力的工具，更希望您的思想和知识通过这个平台，连同优质的阅读体验，将他们分享给有相同志趣的人，进而鼓励更多的人来到这里记录分享他们的思想和知识，尝试点击 (Ctrl+Alt+P) 发布这份文档给好友吧！

再一次感谢您花费时间阅读这份欢迎稿，点击 (Ctrl+Alt+N) 开始撰写新的文稿吧！祝您在这里记录、阅读、分享愉快！

作者 @ghosert
2016 年 07月 07日

^LaTeX. 支持 LaTeX 编辑显示支持，例如：$\sum_{i=1}^n a_i=0$， 访问 MathJax 参考更多使用方法。 ↩

^code. 代码高亮功能支持包括 Java, Python, JavaScript 在内的，四十一种主流编程语言。 ↩