开发: C++知识库 Java知识库 JavaScript Python PHP知识库人工智能区块链大数据移动开发嵌入式开发工具数据结构与算法开发测试游戏开发网络协议系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑笔记本显卡显示器固态硬盘硬盘耳机手机 iphone vivo oppo 小米华为单反装机图拉丁

-> Java知识库 -> RocketMQ底层详解 -> 正文阅读

[Java知识库]RocketMQ底层详解

RocketMQ底层详解

架构

Broker

RocketMQ的服务，或者说是一个进程叫做Broker，作用是存储和转发消息。RocketMQ单机大约能承受10万QPS的请求。为了提升Broker的可用性，以及提升服务器性能，通常会做集群的部署。每个Broker保存总数据的一部分，因此可以实现横向扩展。

为了提升可靠性（防止数据丢失），每个Broker可以拥有自己的副本（slave）。

默认情况，读写都发生在主节点上。在slaveReadEnable = true的情况下，slave也可以参与读负载，但是默认只有BrokerId=1 的slave才会参与，而且是在主节点消费慢的情况下，由参数whichBrokerWhenConsumeSlowly决定，

private long whichBrokerWhenConsumeSlowly = 1;

Topic

用于将消息按照主题做划分，比如订单消息、物流消息。Topic是一个逻辑概念，消息并不是按照Topic划分存储的。

生产者将消息发往指定的Topic,消费者通过订阅这个Topic就可以收到相应的消息。private boolean autoCreateTopicEnable = true设定了如果Topic不存在，就会自动创建。

NameServer

当不同消息存储到不同的Broker上，生产者和消费者对于Broker的读取，由自己实现了一个服务来完成——NameServer。可以理解为RocketMQ的路由中心，自身也可以做集群的部署，作用有点像Eureka和Redis Sentinel。

也就是说，Broker会在NameServer上注册自己，生产者和消费者用NameServer来发现Broker。

NameServer作为路由中心时到底是怎么工作的？

每个Broker节点启动的时候，都会根据配置遍历NameServer列表。

位于：rocketmq/conf/broker.conf

nameserAddr = localhost:9876

与每个NameServer建立TCP长连接，注册自己的信息，之后每隔30s发送心跳信息（服务主动注册）。

如果Broker挂了，不发送心跳了，NameServer怎么发现？

除了主从注册还有定时探活，每隔10s检查一下各个Broker的最近一次心跳时间，如果发现某个Broker超过120s都没有发送心跳，就认为这个Broker已经挂掉，会将其从路由信息里移除。

为什么不使用Zookeeper来作为路由中心？

在之前确实用过，但是后来采用了NameServer。因为架构设计决定了只需要一个轻量级的元数据服务器就够了，只需要保持最终一致性，而不需要Zookeeper这样的强一致性解决方案不需要再依赖里一个中间件，从而减少整体维护的成本。

CAP：C（一致性，Consistency）,A（可用性，Availability）,P（分区容错，Partiton Tolerance）。

Zookeeper 实现了CP，NameServer 选择了AP，放弃了实时一致性。

一致性问题如何解决？

NameServer之间是互相不通信的，也没有主从之分，是如何保持一致性的呢。从三点进行分析：

服务注册：如果新增了Broker，如何新增到所有的NameServer中？因为没有主节点，Broker每隔30s会向所有的NameServer发送心跳信息，所以还是可以保持一致的。
服务剔除：如果一个Broker挂了，怎么从所有的NameServer中移除它的信息？
1. 如果Broker正常关闭：连接断开，Netty的通道关闭监听器会监听到连接断开事件，然后会将这个Broker信息剔除。
2. 如果Broker关闭异常：NameServer的定时任务每10s扫描Broker列表，如果某个Broker的心跳包的最新时间戳超过当前时间120s，就会被移除。
因此，不管是Broker挂了还是恢复了，增加了还是减少了，NameServer都能够保持数据一致。

路由发现：如果Broker的信息更新了（增加或减少节点），客户端如何获得最新的Broker列表？

对于生产者：发送第一条消息的时候，根据Topic从NameServer获取路由信息
对于消费者：消费者一般是订阅固定的Topic，在启动的时候就要获取Broker信息。

在这之后，如果Broker的信息动态变化了怎么办？

因为NameServer不会主动推送服务信息给客户端，客户端也不会发送心跳到Nameserver，所以建立连接之后，需要生产者和消费者定期更新。

在MQClientInstance类（生产者消费者通用）的start方法中，启动了一个定时任务，其中第二个任务：

this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
    @Override
    public void run() {
        try {
            MQClientInstance.this.updateTopicRouteInfoFromNameServer();
        } catch (Exception e) {
            log.error("ScheduledTask updateTopicRouteInfoFromNameServer exception", e);
        }
    }
}, 10, this.clientConfig.getPollNameServerInterval(), TimeUnit.MILLISECONDS);

updateTopicRouteInfoFromNameServer()方法，是用来定期更新NameServer信息的，默认是30s一次。

public void updateTopicRouteInfoFromNameServer() {
    Set<String> topicList = new HashSet<String>();
    // Consumer
    {
        Iterator<Entry<String, MQConsumerInner>> it = this.consumerTable.entrySet().iterator();
        while (it.hasNext()) {
            Entry<String, MQConsumerInner> entry = it.next();
            MQConsumerInner impl = entry.getValue();
            if (impl != null) {
                Set<SubscriptionData> subList = impl.subscriptions();
                if (subList != null) {
                    for (SubscriptionData subData : subList) {
                        topicList.add(subData.getTopic());
                    }
                }
            }
        }
    }
    // Producer
    {
        Iterator<Entry<String, MQProducerInner>> it = this.producerTable.entrySet().iterator();
        while (it.hasNext()) {
            Entry<String, MQProducerInner> entry = it.next();
            MQProducerInner impl = entry.getValue();
            if (impl != null) {
                Set<String> lst = impl.getPublishTopicList();
                topicList.addAll(lst);
            }
        }
    }
    for (String topic : topicList) {
        this.updateTopicRouteInfoFromNameServer(topic);
    }
}

消费者和生产者以相同的时间间隔，更新NameServer信息。拉取的时间间隔由DefaultMQPushConsumer的pollNameServerInterval参数决定，默认是30s。

总结一下：各个NameServer的数据能够保持一致，而且生产者和消费者会定期更新路由信息，所以可以获取最新的消息。

如果Broker刚挂，客户端30s以后才更新路由信息，那是不是会出现最多30s的数据延迟？

解决思路：

重试
把无法连接的Broker进行隔离，不再连接
优先选择延迟小的节点，就能避免连接到容易挂的Broker了。

如果作为路由中心的NameServer全部挂掉了，而且暂时没有恢复呢？

客户端也会缓存Broker信息，所以没有关系，不能完全依赖于NameServer。

Producer

生产者，用于生产消息，会定时从NameServer拉去路由信息（不用配置RocketMQ的服务地址），然后根据路由信息与指定的Broker建立TCP长连接，从而将消息发送到Broker中。发送逻辑一致的Producer可以组成一个Group。

RocketMQ的生产者同样支持批量发送，不过List要自己传进去。Producer写数据只能操作主节点。

Consumer

消息的消费者，通过NameServer集群获得Topic的路由信息，连接到对应的Broker上消费信息。消费逻辑一致的Consumer可以组成一个Group，这时候消息会在Consumer之间负载。由于主节点和从节点都可以读取消息，因此Consumer与所有节点都会建立连接。

一个消费者组中的消费者应该订阅同一个topic。或者反过来，消费不同topic的消费者不应该采用相同的消费者组名字。如果消费了不同的topic，后面的消费者的订阅会覆盖之前消费者的订阅。

两种消费模式：

RocketMQ支持pull和push两种消费模型。

pull

consumer轮询从broker拉取消息。使用DefaultPullConsumer创建实例。

DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("xxxxx");

两种实现方式：

普通的轮询（Polling）：不管服务端数据有无更新，客户端每隔定长时间请求拉去一次数据，可能有更新数据返回，也可能什么都没有。缺点就是大部分时候都没有数据，这些无效的请求会大大地浪费服务器的资源，而且定时请求的间隔过长的时候会导致消息延迟。
长轮询（RocketMQ的pull实现方式）：客户端发起Long Polling，如果此时服务端没有相关数据，会hold住请求，直到服务端有相关数据，或者等待一定时间超时才会返回。返回后，客户端有会立即再次发起下一次长轮询。

轮询唯一的缺点就是服务器在挂起的时候比较消耗内存。

push

这种方式采用Broker将消息推送给Consumer，实现类：DefaultMQPushConsumer

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("xxxxx");

RocketMQ的push模式实际上是基于pull模式实现的，只不过在pull模式上封装了一层，所以Rocketmq push模式并不是真正意义上的“推模式”。

PushConsumer会注册MessageListener监听器，取到消息后，唤醒MessageListener的consumerMessage（）来消费，对用户而言，感觉就像是被推送过来的。

Message Queue

rocketmq支持多主节点的架构。如果发往某一个Topic的多条消息，是不是在所有的Broker上都存储完全相同的内容呢？

肯定不是的。这样不仅浪费了存储空间，还无法通过增加机器数量线性地提升Broker的性能，只能垂直扩展，无法横向水平扩展，性能受限。

那么，如何将一个Topic里面的消息分布到不同的master上呢？

Rocketmq中提供了一个叫做Message Queue的逻辑概念。首先，创建Topic的时候会指定队列的数量，分别是writeQueueNums（写队列数量）和readQueueNums（读队列数量）。

写队列的数量决定了有几个MessageQueue，读队列的数量决定了有几个线程来消费这些MessageQueue（只是用来负载的）。如果没有指定MQ的时候，默认一个Topic拥有8个队列：

private int defaultTopicQueueNums = 8;
//BrokerConfig

topic不存在，生产者发送消息时创建默认4个队列：

private volatile int defaultTopicQueueNums = 4;
//DefaultMQProducer

服务器创建的时候有一个判断，取小一点的值：

int queueNums = Math.min(defaultMQProducer.getDefaultTopicQueueNums(),data.getReadQueueNums());

MQ在磁盘上是可以看到的，但是数量只跟写队列相关，而且在所有主节点上个数是一样的（数据存储不一样）。

发送消息的时候，生产者会根据一定的规则，获得MessageQueue，只要拿到了queueid，就知道要往哪个Broker发送，然后在commitlog写入消息。

RocketMQ 原理

生产者

生产者利用队列可以实现消息的负载和平均分布，那什么时候消息会发到哪个队列呢？

消息发送规则

从Producer的send方法开始跟踪，在DefaultMQProducerImpl的select方法会选择要发送的Queue：

public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
        return this.mqFaultStrategy.selectOneMessageQueue(tpInfo, lastBrokerName);
    }

调用MQFaultStratage的选择队列的方法，这个类是MQ负载均衡的核心类：

int index = tpInfo.getSendWhichQueue().incrementAndGet();
                for (int i = 0; i < tpInfo.getMessageQueueList().size(); i++) {
                    int pos = Math.abs(index++) % tpInfo.getMessageQueueList().size();
                    if (pos < 0)
                        pos = 0;
                    MessageQueue mq = tpInfo.getMessageQueueList().get(pos);
                    if (latencyFaultTolerance.isAvailable(mq.getBrokerName()))
                        return mq;
                }

MessageQueueSelector 有三个实现类：

默认，他是一种不断自增、轮询的方式
随机选择一个队列
返回空，没有实现

也可以自定义MessageQueueSelector，作为参数传进去：

SendResult sendResult = producer.send(msg,new MessageQueueSelector(){
	public MessageQueue select(List<MessageQueue> mqs,Message msg,Object arg){
        Integer id = (Integer)arg;
        int index = id%mqs.size();
        return mqs.get(index);
    }
},i);

顺序消息

原本代码中调用是有序的，但是消息中间件经过了Broker的转发，而且可能出现多个消费者并发消费，就会导致乱序的问题。

**全局有序：**不管有几个生产者，在服务端怎么写入，有几个消费者。消费的顺序和生产顺序是一致的。实现比较麻烦，即使实现了，也会对MQ的性能产生很大的影响。

**顺序消息：**实际上就是局部有序。那么，例如：订单服务。只需要同一个订单相关的消费是有序的就ok了。

保证消息有序，要分成几个环节分析：

生产者发送消息的时候，到达Broker应该是有序的，所以对于生产者，不能使用多线程异步发送，而是顺序发送。
写入Broker的时候，应该是顺序写入的。也就是相同主题的消息应该是集中写入，选择同一个queue，而不是分散写入。
消费者消费的时候只能有一个线程。否则由于消费的速率不同，有可能出现记录到数据库的时候无序。

RocketMQ的顺序消息怎么实现呢？

首先生产者发送消息，是单线程的。追踪send方法发现，消息是同步发送的，最终调用了sendMessageSyn方法同步发送消息。生产者接受到Broker的成功的Response才算是消息发送成功。
其次，消息要路由到相同的queue（关键是要在相同的通道排序，才能实现先进先出）。生产者是如何控制队列的选择？通过MessageQueueSelector，默认使用hashkey进行选择。
```
producer sendOneway(rocketMsg,messageQueueSelector,hashKey); //这边其实时传入了一个hashKey
```
只要传入相同的hashKey，就会选择同一个队列。
最后，在消费者需要保证一个队列只有一个线程消费。Spring Boot中，consumeMode设置为ORDERLY。Java API中，传入MessageListenerOrderly的实现类。
```
consumer.registerMessageListener(new MessageListenerOrderly(){});
//根据注释：一个队列只能有一个线程来消费
```
消费者在启动consumer.start()的时候会判断getMessageListenerInner的实现是哪种（通过instanceof判断），根据Listener类型使用不同的Service，紧接就启动这个Servcie。调用DefaultMQPushConsumerImpl中的pullMessage方法，将拉取到的消息放入ProcessQueue。然后进入ConsumeMessageOrderlyService的submitConsumeRequest方法。拉取到的消息构成了ConsumeRequest（实现了Runnable），然后放入线程池等待执行。
```
@Override    public void submitConsumeRequest(        final List<MessageExt> msgs,        final ProcessQueue processQueue,        final MessageQueue messageQueue,        final boolean dispathToConsume) {        if (dispathToConsume) {            ConsumeRequest consumeRequest = new ConsumeRequest(processQueue, messageQueue);            this.consumeExecutor.submit(consumeRequest);        }    }
```
既然放入了线程池，多线程还能保证有序吗？

ProcessQueue是使用TreeMap来存放消息的：

TreeMap是红黑树的实现，自平衡的排序二叉树。因为Key是当前消息的offset，消息是按照offset排序的。再查看ConsumeRequest的run方法：

在消费的时候加上了锁，RocketMQ顺序消息消费会将队列锁定，当队列获取锁之后才能进行消费，所以能够实现有序消费。

事务消息

**半消息：**暂时不能投递消费者的消息，发送方已经将消息成功发送到了MQ服务器，但是服务端未收到生产者对该消息的二次确认，此时该消息被标记成“暂不能投递”的状态，处于该种状态下的消息即为半消息。

**消息回查： **由于网络闪断、生产者应用重启等原因，导致某条事务消息的二次确认丢失，MQ服务端通过扫描发现某条消息长期处于“半消息”时，需要主动向消息生产者询问该消息的最终状态（Commit或者是Rollback），该过程即消息回查。

整体流程：

**代码实现：**RocketMQ提供了一个TransactionListener接口，这个里面可以实现执行本地事务。

public interface TransactionListener {    /**     * When send transactional prepare(half) message succeed, this method will be invoked to execute local transaction.     *     * @param msg Half(prepare) message     * @param arg Custom business parameter     * @return Transaction state     */    LocalTransactionState executeLocalTransaction(final Message msg, final Object arg);    /**     * When no response to prepare(half) message. broker will send check message to check the transaction status, and this     * method will be invoked to get local transaction status.     *     * @param msg Check message     * @return Transaction state     */    LocalTransactionState checkLocalTransaction(final MessageExt msg);}

默认回查总次数是15次，第一次回查间隔时间是6s，后续每次间隔60s。

延迟消息

RocketMQ可以直接支持延迟消息。但是开源版本功能被阉割了，只能支持特定等级的消息，商业版本可以任意指定时间。比如：

msg.setDelayLevel(3);//level 3 = 10s,延迟级别配置代码在MessageStoreConfig#messageDelayLevel中//springboot中可以这样使用：rocketMQtTemplate.syncSend(topic,message,1000,3);

**实现原理： **Broker中内置延迟消息处理能力，核心实现思路都是一样的：将延迟消息通过一个临时存储进行暂存，到期后才投递到目标Topic中。

步骤：

producer要将一个延迟消息发送到某个Topic中。
Broker判断这是一个延迟消息后，将其通过临时存储进行暂存。
Broker内部通过一个延迟服务（delay Service）检查消息是否到期，将到期的消息投递到目标Topic中。
消费者消费目标Topic中的延迟投递的消息。

临时存储和延迟服务都是在Broker中内部实现，对业务透明。

CommitLog#putMessage：

// Delay Delivery            if (msg.getDelayTimeLevel() > 0) {                if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {                    msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel());                }                topic = TopicValidator.RMQ_SYS_SCHEDULE_TOPIC;                queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());                // Backup real topic, queueId                MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_TOPIC, msg.getTopic());                MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_QUEUE_ID, String.valueOf(msg.getQueueId()));                msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties()));                msg.setTopic(topic);                msg.setQueueId(queueId);            }

消息实际存储是在commitlog，异步写入consume queue。调用ScheduleMessageService#start：

for (Map.Entry<Integer, Long> entry : this.delayLevelTable.entrySet()) {                Integer level = entry.getKey();                Long timeDelay = entry.getValue();                Long offset = this.offsetTable.get(level);                if (null == offset) {                    offset = 0L;                }                if (timeDelay != null) {                    this.timer.schedule(new DeliverDelayedMessageTimerTask(level, offset), FIRST_DELAY_TIME);                }            }

为每个延迟级别都创建一个任务，获得每个级别的延迟时间，并且创建TimerTask。时间到期后，调用了DeliverDelayedMessageTimerTask#messageTimeup(内部类)，其中获取索引之后，读取消息，获取源Topic名称等等，重新将消息持久化到commitlog中并异步写入consume queue。最后消息可以正常消费。

rocketmq的消息重试也是基于延迟消息来完成的。在消费失败的情况下，将其重新当作延迟消息投递回Broker。

Broker

消息存储

既没有分区的概念，也没有按分区存储消息。

为什么这样设计？

每个分区存储整个消息数据。尽管每个分区都是有序写入磁盘的，但随着并发写入分区的数量增加，从操作系统来开，写入就变成了随机的。
由于写入数据文件分散，很难使用批量刷盘

所以RocketMQ设计了一种新的文件存储方式，就是所有的Topic的所有的消息全部写在同一个文件中（这种存储方式叫集中型存储或混合型存储），这样能够保证绝对的顺序写。

优势：

对列轻量化，单个数据队列数据量非常少。
对磁盘的访问串行化，完全顺序写，避免磁盘竞争，不会因为队列增加导致IOWAIT增高。

消费消息的时候，如果想要consumer group 只查找自己的topic的offset消息，可以为每一个consumer group把他们消费的topic的最后消费到的offset单独存储在一个地方。这个存储消息的偏移量的对象就叫做consume queue。

也就是说，消息在Broker存储的时候，不仅写入commitlog，同时也把在commitlog中的最新的offset异步写入对应的consume queue。

消费者在消费消息的时候，先从consume queue读取持久化消息的起始物理位置偏移量offset，大小size和消息tag的hashCode值，随后在从commit log中进行读取待拉取消费消息的真正实体内容部分。

存储关键技术（持久化/硬盘）

CPU如果要读取或者操作磁盘上的数据，必须要把磁盘的数据加载到内存，这个是由硬件结构和访问速度的差异决定的。这个加载的大小有一个固定的单位，叫做Page。x86的linux中一个标准页面大小是4KB。如果要提升磁盘访问速度，或者说尽量减少磁盘I/O，可以把访问过的Page在内存中缓存起来。这个内存的区域就叫做Page Cache。下次处理I/O的时候，就先到Page Cache里面查找，找到了就直接操作。没找到就到磁盘里面查找。

Page Cache本身也会对数据文件进行预读取，对于每个文件的第一个请求操作，系统就在读入所请求页面的同时读入紧跟其后的少数几个页面。

因为虚拟内存分为内核空间和用户空间。Page Cache属于内核空间，用户空间访问不了，因此读取数据据还需要从内核空间拷贝到用户空间缓冲区。这样会降低数据访问的速度。因此在这里使用了一种零拷贝技术。

将Page Cache的数据在用户空间中做一个地址映射，这样用户进程就可以通过指针操作直接读写Page Cache，不需要再使用系统调用和内存拷贝。

在RocketMQ中具体的实现是使用mmap（memory map,内存映射），不论是commitlog 还是 ConsumerQueue都采用了mmap。

文件清理策略

RocketMQ中被消费过的消息是不会删除的，所以保证了文件的顺序写入。因此需要进行文件清理。

需要清理的文件应该满足这几个条件：

是Commitlog 或者是 ConsumeQueue中的文件
超过72小时的文件（才算是过期文件）

那么什么时候删除呢？（分两种情况）

通过定时任务，每天凌晨4点，删除这些过期的文件
磁盘使用空间超过了75%，开始删除过期文件。

当然如果磁盘使用率超过了85%，会开始批量清理文件，不管有没有过期，直到空间充足。如果磁盘空间使用率超过了90%，会拒绝消息的写入。

消费者

如果我们需要提高消费者的负载能力，必然要增加消费者的数量。那么如果做到尽量平均的消费消息？队列如何分配给对应的消费者？

消费端的负载均衡与rebalance

消费者增加的时候，肯定会引起rebalance。在消费者启动的入口函数中调用了RebalanceService中的run方法。也就是说：

消费者启动的时候或者由消费者挂掉的时候，默认最多20秒（实际上在DefaultMQPushConsumerImpl的start方法末尾还有一句：this.mQClientFactory.rebalanceImmediately();会唤醒沉睡的线程，也就是立即执行RebalanceService的run方法），就会做一次rebalance，让所有的消费者可以尽量均匀的消费队列消息。

具体是如何进行rebalance的呢？

一共存在六种已经实现的策略（可以指定，也可以通过consumer.setAllocateMessageQueueStrategy();自定义实现）：

AllocateMessageQueueAveragely: 连续分配（默认）
AllocateMessageQueueAveragelyByCircle: 每人轮流一个
AllocateMessageQueueByConfig: 通过配置
AllocateMessageQueueConsistentHash: 一致性哈希
AllocateMessageQueueByMachineRoom: 指定一个Broker的topic中的queue消费
AllocateMachineRoomNearby: 按Broker的机房就近分配

队列的数量尽量大于消费者数量。

消费端重试与死信队列

如果一条消息消费失败，返还给Broker的是RECONSUME_LATER，表示稍后重试。这个时候，消息会发回到Broker，进入到RocketMQ的重试队列中，服务端会为consumer group创建一个名字为%RETRY%开头的重试队列。

重试队列的消息会过一段时间再次发送给消费者，如果还是异常，会再次进入重试队列，重试的时间间隔不会不断衰减，从10秒开始直到2个小时最多重试16次：10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h。（同样也是延迟消息的时间等级，从level = 3开始。也就是说重试队列使用延迟队列的功能实现的）发到对应的SCHEDULE_TOPIC_XXXX，到时间后再替换成真实的topic，实现重试。

那么如果重试了16次都没有成功，这个时候消息就会丢到死信队列中。Broker会创建一个死信队列，名字是：%DLQ%+ConsumerGroupName。死信队列中的消息最后需要人工处理，可以写一个线程进行对死信队列的订阅，人工消费消息。

高可用架构

在高可用架构中，我们主要关注两个模块：主从同步和故障转移。

数据同步

主从的关联

主从服务器如何联系在一起？

集群的名字设置为相同，borkerClusterName = xxxx-cluster
连接到相同的NameServer。
在配置文件中配置：brokerId = 0 代表是主节点，brokerId = 1 代表是slave。

主从同步和刷盘类型

在部署节点的时候，配置文件中设置了Broker角色和刷盘方式：

通常情况下，会把主节点和从节点的Broker配置成ASYNC_FLUSH异步刷盘的方式，主从之间配置成SYNC_MASTER同步复制的方式，即：异步刷盘+同步复制。

下图是异步刷盘和同步刷盘的理解图：

主从同步的流程

从服务器主动建立TCP连接主服务器，然后每隔5s向主服务器发送commitLog文件最大偏移量拉取还未同步的消息。
主服务器开启监听端口，监听从服务器发送过来的信息，主服务器收到从服务器发过来的偏移量进行解析，并返回查找出未同步的消息给从服务器。
客户端收到主服务器的消息后，将这批消息写入commitLog文件中，然后更新commitLog拉取偏移量，接着继续向主服务器拉取未同步消息。

高可用与故障转移

在之前的版本中，RocketMQ只有一种master/slave部署方式：一组Broker中有一个master和零个到多个slave，slave通过同步复制或异步复制的方式去同步master的数据。提供了一定的高可用性。

但是这样的部署模式具有一定的缺陷。比如故障转移方面：如果主节点挂了还需要人为手动的进行重启或者切换。

在4.5.0的版本中，rocketmq利用了Dledger技术解决了自动选主的问题。引入Dledger技术，其实就是使用Dledger的CommitLog来替换掉Broker自己的CommitLog。然后Broker仍然可以基于Dledger的CommitLog，把消息的位置信息保存到ConsumeQueue中。

Dledger是一个基于raft协议的commitlog存储库，也是rocketmq实现新的高可用多副本架构的关键。优点是：不需要外部引入组件，自动选主逻辑集成到各个节点的进程中，节点之间通过通信就可以完成选主。

在这种情况下，commitlog是Dledger管理的，具有选主的功能。

#是否启用Dledger,即是否启用主从切换，默认值为falseenableDLegerCommitLog = true#DLedger Raft Group 的名字dLegerGroup = broker-a#DLedger Group 内各节点的地址和端口，至少需要3个节点dLegerPeers = n0-192.168.44.163:10911;n1-[server-ip]:[server:port];n[number]-[server-ip]:[server-port]#本节点IDdLegerSelfId = n0