RocketMQ 面试题

1、说说你们公司线上生产环境用的是什么消息中间件?

见【2、多个 mq 如何选型？】

2、多个 mq 如何选型？

MQ	描述
RabbitMQ	erlang开发，对消息堆积的支持并不好，当大量消息积压的时候，会导致 RabbitMQ 的性能急剧下降。每秒钟可以处理几万到十几万条消息。
RocketMQ	java开发，面向互联网集群化功能丰富，对在线业务的响应时延做了很多的优化，大多数情况下可以做到毫秒级的响应，每秒钟大概能处理几十万条消息。
Kafka	Scala开发，面向日志功能丰富，性能最高。当你的业务场景中，每秒钟消息数量没有那么多的时候，Kafka 的时延反而会比较高。所以，Kafka 不太适合在线业务场景。
ActiveMQ	java开发，简单，稳定，性能不如前面三个。小型系统用也ok，但是不推荐。推荐用互联网主流的。

3、为什么要使用 MQ？

因为项目比较大，做了分布式系统，所有远程服务调用请求都是同步执行经常出问题，所以引入了 mq

作用	描述
解耦	系统耦合度降低，没有强依赖关系
异步	不需要同步执行的远程调用可以有效提高响应时间
削峰	请求达到峰值后，后端service还可以保持固定消费速率消费，不会被压垮

4、RocketMQ 由哪些角色组成，每个角色作用和特点是什么？

角色	作用
Nameserver	无状态，动态列表；这也是和zookeeper的重要区别之一。zookeeper是有状态的。
Producer	消息生产者，负责发消息到Broker。
Broker	就是MQ本身，负责收发消息、持久化消息等。
Consumer	消息消费者，负责从Broker上拉取消息进行消费，消费完进行ack。

5、RocketMQ 中的 Topic 和 JMS 的 queue 有什么区别？

queue 就是来源于数据结构的 FIFO 队列。而 Topic 是个抽象的概念，每个 Topic 底层对应 N 个 queue，而数据也真实存在 queue 上的。

6、RocketMQ Broker 中的消息被消费后会立即删除吗？

不会，每条消息都会持久化到 CommitLog 中，每个 Consumer 连接到 Broker 后会维持消费进度信息，当有消息消费后只是当前 Consumer 的消费进度（CommitLog 的 offset）更新了。

追问：那么消息会堆积吗？什么时候清理过期消息？

4、 6 版本默认 48 小时后会删除不再使用的 CommitLog 文件；

• 检查这个文件最后访问时间
• 判断是否大于过期时间
• 指定时间删除，默认凌晨 4 点

源码如下：

/**
 * {@link org.apache.rocketmq.store.DefaultMessageStore.CleanCommitLogService#isTimeToDelete()}
 */
private boolean isTimeToDelete() {


    // when = "04";
    String when = DefaultMessageStore.this.getMessageStoreConfig().getDeleteWhen();
    // 是04点，就返回true
    if (UtilAll.isItTimeToDo(when)) {


        return true;
    }
	// 不是04点，返回false
    return false;
}

/**
 * {@link org.apache.rocketmq.store.DefaultMessageStore.CleanCommitLogService#deleteExpiredFiles()}
 */
private void deleteExpiredFiles() {


    // isTimeToDelete()这个方法是判断是不是凌晨四点，是的话就执行删除逻辑。
    if (isTimeToDelete()) {


        // 默认是72，但是broker配置文件默认改成了48，所以新版本都是48。
        long fileReservedTime = 48 * 60 * 60 * 1000;
        deleteCount = DefaultMessageStore.this.commitLog.deleteExpiredFile(72 * 60 * 60 * 1000, xx, xx, xx);
    }
}

/**
 * {@link org.apache.rocketmq.store.CommitLog#deleteExpiredFile()}
 */
public int deleteExpiredFile(xxx) {


    // 这个方法的主逻辑就是遍历查找最后更改时间+过期时间，小于当前系统时间的话就删了（也就是小于48小时）。
    return this.mappedFileQueue.deleteExpiredFileByTime(72 * 60 * 60 * 1000, xx, xx, xx);
}

7、RocketMQ 消费模式有几种？

消费模型由 Consumer 决定，消费维度为 Topic。

• 集群消费

1.一条消息只会被同 Group 中的一个 Consumer 消费

2.多个 Group 同时消费一个 Topic 时，每个 Group 都会有一个 Consumer 消费到数据

• 广播消费

消息将对一 个 Consumer Group 下的各个 Consumer 实例都消费一遍。即即使这些 Consumer 属于同一个 Consumer Group ，消息也会被 Consumer Group 中的每个 Consumer 都消费一次。

8、消费消息是 push 还是 pull？

RocketMQ 没有真正意义的 push，都是 pull，虽然有 push 类，但实际底层实现采用的是长轮询机制，即拉取方式

broker 端属性 longPollingEnable 标记是否开启长轮询。默认开启

源码如下：

// {@link org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl#pullMessage()}
// 看到没，这是一只披着羊皮的狼，名字叫PushConsumerImpl，实际干的确是pull的活。

// 拉取消息，结果放到pullCallback里
this.pullAPIWrapper.pullKernelImpl(pullCallback);

追问：为什么要主动拉取消息而不使用事件监听方式？

事件驱动方式是建立好长连接，由事件（发送数据）的方式来实时推送。

如果 broker 主动推送消息的话有可能 push 速度快，消费速度慢的情况，那么就会造成消息在 consumer 端堆积过多，同时又不能被其他 consumer 消费的情况。而 pull 的方式可以根据当前自身情况来 pull，不会造成过多的压力而造成瓶颈。所以采取了 pull 的方式。

9、broker 如何处理拉取请求的？

Consumer 首次请求 Broker

• Broker 中是否有符合条件的消息

• 有 ->

• 响应 Consumer

• 等待下次 Consumer 的请求

• 没有

• 挂起 consumer 的请求，即不断开连接，也不返回数据

• 使用 consumer 的 offset，

• DefaultMessageStore#ReputMessageService#run 方法

• 每隔 1ms 检查 commitLog 中是否有新消息，有的话写入到 pullRequestTable

• 当有新消息的时候返回请求

• PullRequestHoldService 来 Hold 连接，每个 5s 执行一次检查 pullRequestTable 有没有消息，有的话立即推送

10、RocketMQ 如何做负载均衡？

通过 Topic 在多 Broker 中分布式存储实现。

producer 端

发送端指定 message queue 发送消息到相应的 broker，来达到写入时的负载均衡

• 提升写入吞吐量，当多个 producer 同时向一个 broker 写入数据的时候，性能会下降
• 消息分布在多 broker 中，为负载消费做准备

默认策略是随机选择：

• producer 维护一个 index
• 每次取节点会自增
• index 向所有 broker 个数取余
• 自带容错策略

其他实现：

• SelectMessageQueueByHash

• hash 的是传入的 args

• SelectMessageQueueByRandom

• SelectMessageQueueByMachineRoom 没有实现

也可以自定义实现MessageQueueSelector接口中的 select 方法

MessageQueue select(final List<MessageQueue> mqs, final Message msg, final Object arg);

consumer 端

采用的是平均分配算法来进行负载均衡。

其他负载均衡算法

平均分配策略(默认)(AllocateMessageQueueAveragely)
环形分配策略(AllocateMessageQueueAveragelyByCircle)
手动配置分配策略(AllocateMessageQueueByConfig)
机房分配策略(AllocateMessageQueueByMachineRoom)
一致性哈希分配策略(AllocateMessageQueueConsistentHash)
靠近机房策略(AllocateMachineRoomNearby)

追问：当消费负载均衡 consumer 和 queue 不对等的时候会发生什么？

Consumer 和 queue 会优先平均分配，如果 Consumer 少于 queue 的个数，则会存在部分 Consumer 消费多个 queue 的情况，如果 Consumer 等于 queue 的个数，那就是一个 Consumer 消费一个 queue，如果 Consumer 个数大于 queue 的个数，那么会有部分 Consumer 空余出来，白白的浪费了。

11、消息重复消费

影响消息正常发送和消费的重要原因是网络的不确定性。

引起重复消费的原因

• ACK

正常情况下在 consumer 真正消费完消息后应该发送 ack，通知 broker 该消息已正常消费，从 queue 中剔除

当 ack 因为网络原因无法发送到 broker，broker 会认为词条消息没有被消费，此后会开启消息重投机制把消息再次投递到 consumer

• 消费模式

在 CLUSTERING 模式下，消息在 broker 中会保证相同 group 的 consumer 消费一次，但是针对不同 group 的 consumer 会推送多次

解决方案

• 数据库表

处理消息前，使用消息主键在表中带有约束的字段中 insert

• Map

单机时可以使用 map ConcurrentHashMap -> putIfAbsent guava cache

• Redis

分布式锁搞起来。

12、如何让 RocketMQ 保证消息的顺序消费

你们线上业务用消息中间件的时候，是否需要保证消息的顺序性?

如果不需要保证消息顺序，为什么不需要?假如我有一个场景要保证消息的顺序，你们应该如何保证?

首先多个 queue 只能保证单个 queue 里的顺序，queue 是典型的 FIFO，天然顺序。多个 queue 同时消费是无法绝对保证消息的有序性的。所以总结如下：

同一 topic，同一个 QUEUE，发消息的时候一个线程去发送消息，消费的时候 一个线程去消费一个 queue 里的消息。

追问：怎么保证消息发到同一个 queue？

Rocket MQ 给我们提供了 MessageQueueSelector 接口，可以自己重写里面的接口，实现自己的算法，举个最简单的例子：判断i % 2 == 0，那就都放到 queue1 里，否则放到 queue2 里。

for (int i = 0; i < 5; i++) {


    Message message = new Message("orderTopic", ("hello!" + i).getBytes());
    producer.send(
        // 要发的那条消息
        message,
        // queue 选择器 ，向 topic中的哪个queue去写消息
        new MessageQueueSelector() {


            // 手动 选择一个queue
            @Override
            public MessageQueue select(
                // 当前topic 里面包含的所有queue
                List<MessageQueue> mqs,
                // 具体要发的那条消息
                Message msg,
                // 对应到 send（） 里的 args，也就是2000前面的那个0
                Object arg) {


                // 向固定的一个queue里写消息，比如这里就是向第一个queue里写消息
                if (Integer.parseInt(arg.toString()) % 2 == 0) {


                    return mqs.get(0);
                } else {


                    return mqs.get(1);
                }
            }
        },
        // 自定义参数：0
        // 2000代表2000毫秒超时时间
        i, 2000);
}

13、RocketMQ 如何保证消息不丢失

首先在如下三个部分都可能会出现丢失消息的情况：

• Producer 端
• Broker 端
• Consumer 端

13.1、Producer 端如何保证消息不丢失

• 采取 send()同步发消息，发送结果是同步感知的。
• 发送失败后可以重试，设置重试次数。默认 3 次。

producer.setRetryTimesWhenSendFailed(10);

• 集群部署，比如发送失败了的原因可能是当前 Broker 宕机了，重试的时候会发送到其他 Broker 上。

13.2、Broker 端如何保证消息不丢失

• 修改刷盘策略为同步刷盘。默认情况下是异步刷盘的。

flushDiskType = SYNC_FLUSH

• 集群部署，主从模式，高可用。

13.3、Consumer 端如何保证消息不丢失

• 完全消费正常后在进行手动 ack 确认。

14、rocketMQ 的消息堆积如何处理

下游消费系统如果宕机了，导致几百万条消息在消息中间件里积压，此时怎么处理?

你们线上是否遇到过消息积压的生产故障?如果没遇到过，你考虑一下如何应对?

首先要找到是什么原因导致的消息堆积，是 Producer 太多了，Consumer 太少了导致的还是说其他情况，总之先定位问题。

然后看下消息消费速度是否正常，正常的话，可以通过上线更多 consumer 临时解决消息堆积问题

追问：如果 Consumer 和 Queue 不对等，上线了多台也在短时间内无法消费完堆积的消息怎么办？

• 准备一个临时的 topic
• queue 的数量是堆积的几倍
• queue 分布到多 Broker 中
• 上线一台 Consumer 做消息的搬运工，把原来 Topic 中的消息挪到新的 Topic 里，不做业务逻辑处理，只是挪过去
• 上线 N 台 Consumer 同时消费临时 Topic 中的数据
• 改 bug
• 恢复原来的 Consumer，继续消费之前的 Topic

追问：堆积时间过长消息超时了？

RocketMQ 中的消息只会在 commitLog 被删除的时候才会消失，不会超时。也就是说未被消费的消息不会存在超时删除这情况。

追问：堆积的消息会不会进死信队列？

不会，消息在消费失败后会进入重试队列（%RETRY%+ConsumerGroup），16 次（默认 16 次）才会进入死信队列（%DLQ%+ConsumerGroup）。

源码如下：

public class SubscriptionGroupConfig {


    private int retryMaxTimes = 16;
}

// {@link org.apache.rocketmq.broker.processor.SendMessageProcessor#asyncConsumerSendMsgBack}
// maxReconsumeTimes = 16
int maxReconsumeTimes = subscriptionGroupConfig.getRetryMaxTimes();
// 如果重试次数大于等于16，则创建死信队列
if (msgExt.getReconsumeTimes() >= maxReconsumeTimes || delayLevel < 0) {


    // MixAll.getDLQTopic()就是给原有groupname拼上DLQ，死信队列
    newTopic = MixAll.getDLQTopic(requestHeader.getGroup());
    // 创建死信队列
    topicConfig = this.brokerController.getTopicConfigManager().createTopicInSendMessageBackMethod(xxx)
}

扩展：每次重试的时间间隔：

public class MessageStoreConfig {


    // 每隔如下时间会进行重试，到最后一次时间重试失败的话就进入死信队列了。
	private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
}

看到这个源码你可能蒙蔽了，这不是 18 个时间间隔嘛。怎么是 16 次？继续看下面代码，我 TM 也懵了。

/**
 * {@link org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl#sendMessageBack()}
 *
 * sendMessageBack()这个方法是消费失败后会请求他，意思是把消息重新放到队列，进行重试。
 */
public void sendMessageBack(MessageExt msg, int delayLevel, final String brokerName) {


    Message newMsg = new Message();
    // ！！！我TM，真相了，3 + xxx。他是从第三个开始的。也就是舍弃了前两个时间间隔，18 - 2 = 16。也就是说第一次重试是在10s，第二次30s。
    // TMD！！！
    // TMD！！！
    // TMD！！！
    newMsg.setDelayTimeLevel(3 + msg.getReconsumeTimes());
	this.mQClientFactory.getDefaultMQProducer().send(newMsg);
 }

15、RocketMQ 在分布式事务支持这块机制的底层原理?

你们用的是 RocketMQ?RocketMQ 很大的一个特点是对分布式事务的支持，你说说他在分布式事务支持这块机制的底层原理?

分布式系统中的事务可以使用 TCC（Try、Confirm、Cancel）、2pc 来解决分布式系统中的消息原子性

RocketMQ 4.3+提供分布事务功能，通过 RocketMQ 事务消息能达到分布式事务的最终一致

RocketMQ 实现方式：

Half Message： 预处理消息，当 broker 收到此类消息后，会存储到 RMQ_SYS_TRANS_HALF_TOPIC 的消息消费队列中

检查事务状态： Broker 会开启一个定时任务，消费 RMQ_SYS_TRANS_HALF_TOPIC 队列中的消息，每次执行任务会向消息发送者确认事务执行状态（提交、回滚、未知），如果是未知，Broker 会定时去回调在重新检查。

超时： 如果超过回查次数，默认回滚消息。

也就是他并未真正进入 Topic 的 queue，而是用了临时 queue 来放所谓的 half message，等提交事务后才会真正的将 half message 转移到 topic 下的 queue。

16、如果让你来动手实现一个分布式消息中间件，整体架构你会如何设计实现?

我个人觉得从以下几个点回答吧：

• 需要考虑能快速扩容、天然支持集群
• 持久化的姿势
• 高可用性
• 数据 0 丢失的考虑
• 服务端部署简单、client 端使用简单

17、看过 RocketMQ 的源码没有。如果看过，说说你对 RocketMQ 源码的理解?

要真让我说，我会吐槽蛮烂的，首先没任何注释，可能是之前阿里巴巴写了中文注释，捐赠给 apache 后，apache 觉得中文注释不能留，自己又懒得写英文注释，就都给删了。里面比较典型的设计模式有单例、工厂、策略、门面模式。单例工厂无处不在，策略印象深刻比如发消息和消费消息的时候 queue 的负载均衡就是 N 个策略算法类，有随机、hash 等，这也是能够快速扩容天然支持集群的必要原因之一。持久化做的也比较完善，采取的 CommitLog 来落盘，同步异步两种方式。

18、高吞吐量下如何优化生产者和消费者的性能?

开发

• 同一 group 下，多机部署，并行消费

• 单个 Consumer 提高消费线程个数

• 批量消费

• 消息批量拉取

• 业务逻辑批量处理

运维

• 网卡调优
• jvm 调优
• 多线程与 cpu 调优
• Page Cache

19、再说说 RocketMQ 是如何保证数据的高容错性的?

• 在不开启容错的情况下，轮询队列进行发送，如果失败了，重试的时候过滤失败的 Broker
• 如果开启了容错策略，会通过 RocketMQ 的预测机制来预测一个 Broker 是否可用
• 如果上次失败的 Broker 可用那么还是会选择该 Broker 的队列
• 如果上述情况失败，则随机选择一个进行发送
• 在发送消息的时候会记录一下调用的时间与是否报错，根据该时间去预测 broker 的可用时间

其实就是 send 消息的时候 queue 的选择。源码在如下：

org.apache.rocketmq.client.latency.MQFaultStrategy#selectOneMessageQueue()

20、任何一台 Broker 突然宕机了怎么办？

Broker 主从架构以及多副本策略。Master 收到消息后会同步给 Slave，这样一条消息就不止一份了，Master 宕机了还有 slave 中的消息可用，保证了 MQ 的可靠性和高可用性。而且 Rocket MQ4.5.0 开始就支持了 Dlegder 模式，基于 raft 的，做到了真正意义的 HA。

21、Broker 把自己的信息注册到哪个 NameServer 上？

这么问明显在坑你，因为 Broker 会向所有的 NameServer 上注册自己的信息，而不是某一个，是每一个，全部！