阿里二面：RocketMQ 消息积压了，增加消费者有用吗？

面试官RocketMQ 消息积压了增加消费者有用吗我这个要看具体的场景不同的场景下情况是不一样的。面试官可以详细说一下吗我如果消费者的数量小于 MessageQueue 的数量增加消费者可以加快消息消费速度减少消息积压。比如一个 Topic 有 4 个 MessageQueue2 个消费者进行消费如果增加一个消费者明细可以加快拉取消息的频率。如下图如果消费者的数量大于等于 MessageQueue 的数量增加消费者是没有用的。比如一个 Topic 有 4 个 MessageQueue并且有 4 个消费者进行消费。如下图面试官你说的第一种情况增加消费者一定能加快消息消费的速度吗我这...一般情况下是可以的。面试官有特殊的情况吗我当然有。消费者消息拉取的速度也取决于本地消息的消费速度如果本地消息消费的慢就会延迟一段时间后再去拉取。面试官在什么情况下消费者会延迟一段时间后后再去拉取呢我消费者拉取的消息存在 ProcessQueue消费者是有流量控制的如果出现下面三种情况就不会主动去拉取ProcessQueue 保存的消息数量超过阈值默认 1000可以配置ProcessQueue 保存的消息大小超过阈值默认 100M可以配置对于非顺序消费的场景ProcessQueue 中保存的最后一条和第一条消息偏移量之差超过阈值默认 2000可以配置。这部分源码请参考类org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl。面试官还有其他情况吗我对于顺序消费的场景ProcessQueue 加锁失败也会延迟拉取这个延迟时间是 3s。面试官消费者延迟拉取消息一般可能是什么原因导致的呢我其实延迟拉取的本质就是消费者消费慢导致下次去拉取的时候 ProcessQueue 中积压的消息超过阈值。以下面这张架构图为例消费者消费慢可是能下面的原因消费者处理的业务逻辑复杂耗时很长消费者有慢查询或者数据库负载高导致响应慢缓存等中间件响应慢比如 Redis 响应慢调用外部服务接口响应慢。面试官对于外部接口响应慢的情况有什么应对措施吗我这个要分情况讨论。如果调用外部系统只是一个通知或者调用外部接口的结果并不处理可以采用异步的方式异步逻辑里采用重试的方式保证接口调成功。如果外部接口返回结果必须要处理可以考虑接口返回的结果是否可以缓存默认值要考虑业务可行在调用失败后采用快速降级的方式使用默认值替代返回接口返回值。如果这个接口返回结果必须要处理并且不能缓存可以把拉取到的消息存入本地然后给 Broker 直接返回 CONSUME_SUCCESS。等外部系统恢复正常后再从本地取出来进行处理。面试官如果消费者数小于 MessageQueue 数量并且外部系统响应正常为了快速消费积压消息而增加消费者有什么需要考虑的吗我外部系统虽然响应正常但是增加多个消费者后外部系统的接口调用量会突增如果达到吞吐量上限外部系统会响应变慢甚至被打挂。同时也要考虑本地数据库、缓存的压力如果数据库响应变慢处理消息的速度就会变慢起不到缓解消息积压的作用。面试官新增加了消费者后怎么给它分配 MessageQueue 呢我Consumer 在拉取消息之前需要对 MessageQueue 进行负载操作。RocketMQ 使用一个定时器来完成负载操作默认每间隔 20s 重新负载一次。面试官能详细说一下都有哪些负载策略吗我RocketMQ 提供了 6 种负载策略依次来看一下。平均负载策略把消费者进行排序计算每个消费者可以平均分配的 MessageQueue 数量如果消费者数量大于 MessageQueue 数量多出的消费者就分不到如果不可以平分就使用 MessageQueue 总数量对消费者数量求余数 mod对前 mod 数量消费者每个消费者加一个这样就获取到了每个消费者分配的 MessageQueue 数量。比如 4 个 MessageQueue 和 3 个消费者的情况源代码的逻辑非常简单如下// AllocateMessageQueueAveragely 这个类 // 4 个 MessageQueue 和 3 个消费者的情况假如第一个index 0 int index cidAll.indexOf(currentCID); // mod 1 int mod mqAll.size() % cidAll.size(); // averageSize 2 int averageSize mqAll.size() cidAll.size() ? 1 : (mod 0 index mod ? mqAll.size() / cidAll.size() 1 : mqAll.size() / cidAll.size()); // startIndex 0 int startIndex (mod 0 index mod) ? index * averageSize : index * averageSize mod; // range 2,所以第一个消费者分配到了2个 int range Math.min(averageSize, mqAll.size() - startIndex); for (int i 0; i range; i) { result.add(mqAll.get((startIndex i) % mqAll.size())); }循环分配策略:这个很容易理解遍历消费者把 MessageQueue 分一个给遍历到的消费者如果 MessageQueue 数量比消费者多需要进行多次遍历遍历次数等于 MessageQueue 数量/消费者数量还是以 4 个 MessageQueue 和 3 个消费者的情况如下图源代码如下://AllocateMessageQueueAveragelyByCircle 这个类 //4 个 MessageQueue 和 3 个消费者的情况假如第一个index 0 int index cidAll.indexOf(currentCID); for (int i index; i mqAll.size(); i) { if (i % cidAll.size() index) { //i 0 或者 i 3 都会走到这里 result.add(mqAll.get(i)); } }自定义分配策略这种策略在消费者启动的时候可以指定消费哪些 MessageQueue。可以参考下面代码AllocateMessageQueueByConfig allocateMessageQueueByConfig new AllocateMessageQueueByConfig(); //绑定消费 messageQueue1 allocateMessageQueueByConfig.setMessageQueueList(Arrays.asList(new MessageQueue(messageQueue1,broker1,0))); consumer.setAllocateMessageQueueStrategy(allocateMessageQueueByConfig); consumer.start();按照机房分配策略这种方式 Consumer 只消费指定机房的 MessageQueue如下图Consumer0、Consumer1、Consumer2 绑定 room1 和 room2 这两个机房而 room3 这个机房没有消费者。Consumer 启动的时候需要绑定机房名称。可以参考下面代码AllocateMessageQueueByMachineRoom allocateMessageQueueByMachineRoom new AllocateMessageQueueByMachineRoom(); //绑定消费 room1 和 room2 这两个机房 allocateMessageQueueByMachineRoom.setConsumeridcs(new HashSet(Arrays.asList(room1,room2))); consumer.setAllocateMessageQueueStrategy(allocateMessageQueueByMachineRoom); consumer.start();这种策略 broker 的命名必须按照格式机房名brokerName因为消费者分配队列的时候首先按照机房名称过滤出所有的 MessageQueue然后再按照平均分配策略进行分配。//AllocateMessageQueueByMachineRoom 这个类 ListMessageQueue premqAll new ArrayListMessageQueue(); for (MessageQueue mq : mqAll) { String[] temp mq.getBrokerName().split(); if (temp.length 2 consumeridcs.contains(temp[0])) { premqAll.add(mq); } } //上面按照机房名称过滤出所有的 MessageQueue 放入premqAll后面就是平均分配策略按照机房就近分配跟按照机房分配原则相比就近分配的好处是可以对没有消费者的机房进行分配。如下图机房 3 的 MessageQueue 也分配到了消费者如果一个机房没有消费者则会把这个机房的 MessageQueue 分配给集群中所有的消费者。源码所在类AllocateMachineRoomNearby。一致性 Hash 算法策略把所有的消费者经过 Hash 计算分布到 Hash 环上对所有的 MessageQueue 进行 Hash 计算找到顺时针方向最近的消费者节点进行绑定。如下图源代码如下//所在类 AllocateMessageQueueConsistentHash CollectionClientNode cidNodes new ArrayListClientNode(); for (String cid : cidAll) { cidNodes.add(new ClientNode(cid)); } //使用消费者构建 Hash 环把消费者分布在 Hash 环节点上 final ConsistentHashRouterClientNode router; //for building hash ring if (customHashFunction ! null) { router new ConsistentHashRouterClientNode(cidNodes, virtualNodeCnt, customHashFunction); } else { router new ConsistentHashRouterClientNode(cidNodes, virtualNodeCnt); } //对 MessageQueue 做 Hash 运算找到环上距离最近的消费者 ListMessageQueue results new ArrayListMessageQueue(); for (MessageQueue mq : mqAll) { ClientNode clientNode router.routeNode(mq.toString()); if (clientNode ! null currentCID.equals(clientNode.getKey())) { results.add(mq); } }面试官恭喜你通过了。2026年最新版百万八股面试宝典《Java》《大模型》《大并发》《大数据》本题库由资深讲师联合大厂面试官精心打造专为高级IT岗位设计【点击领取-百万八股面试宝典】面试突击体系相关Java企业架构体系相关AI大模型相关HR 面试软技能设计模式相关并发编程相关网络IO与Netty相关互联网三高项目相关亿级流量多级缓存相关数据结构算法相关分布式相关分布式锁相关分布式ID相关核心源码相关大厂线上故障分析相关Docker相关Dubbo相关ElasticSearch相关GoLang相关Java基础核心JVM核心相关JVM调优底层相关Kafka相关Kubernetes相关Linux相关MongoDB相关MQ相关MySQL相关MyBatis相关MyBatisPlus相关Postgresql相关RabbitMQ相关RocketMQ相关Redis相关ShardingSphere相关Shiro相关SkyWalking相关Solr相关Spring 相关SpringBoot相关SpringCloud相关SpringMVC源码相关SpringSecurity相关Spring WebFlux相关领域驱动设计DDD相关任务调度Airflow相关Zookeeper相关HDFS面试题MapReduce面试题Yarn面试题Hive相关SQL面试题相关Scala面试题Spark面试题Flink相关Doris相关Tomcat源码相关网络安全相关运维/云原生相关