消息中间件面试常见问题

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

为什么要使用MQ？

多个MQ如何选型？

使用的Rocket MQ 。
因为项目比较大，做了分布式系统，所有远程服务调用请求都是同步调用，经常出问题，所以引入MQ（解耦）。批量操作需要异步操作，项目中批量打印快递单，批量发货。（解耦，异步，削峰）
用的RocketMQ,1.用java开发的，易于开发。2.能够应对大流量场景，经过双十一的检验
各种MQ对比，小项目：ActiveMQ 大项目：RocketMQ 或者Kafka ，RabbitMQ 。RabbitMQ ：使用erlang开发，延迟低。Kafka ：Scala 开发，面向日志功能丰富。ActiveMQ ：java开发，简单，稳定。

Rocket MQ由哪些角色组成，每个角色的作用和特点是什么？

Rocket MQ 中Topic 和ActiveMQ有什么区别？

Rocket MQ Broker中的消息被消费后会立即删除吗？

生产者、消费者、Broker、NameServer NameServer ：注册中心，无状态：启动后不与其他节点有通信，当作路由，一个broker启动后会向所有NameServer发送注册信息，NameServer维护一个动态列表。
ActiveMQ ：中有destination的概念，即消息目的地，destination分为两类， topic 广播消息，queue 队列消息
RocketMQ ：是一组 MessageMQ的集合　，一条消息是广播消息还是队列消息由客户端消费决定

RocketMQ消费模式有几种？

消费消息时使用的是push还是pull？

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

说一说几种常用的消息同步机制？

集群消费

一组为单位，一组cosumer同时消费一个topic，多个组组成一个集群。

一组consumer 同时消费一个topic，可以分配消费负载均衡策略分配consumer对应消费topic下的哪些queue。

多个group同时消费一个topic时，每个group 都会消费到数据

一条消息只会被一个group中的consumer消费

广播消费

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

Broker 中的消息被消费后不会被立即删除，每条消息都会持久化到CommitLog中，每个consumer连接到broker后会维持消费进度信息，当消息消费后只是当前consumer的消费进度（commitLog 的 offeset）更新了。

在刚开始的时候就要决定使用那种方式消费

两种：DefaultLitePullConsumerImpl 拉 DefaultMQPushConsumerImpl 推

两个都实现了MQConsumerInner 接口，实际底层都实现了长轮询机制，即拉取方式

RocketMQ 消息重试机制

转自：https://blog.csdn.net/belongtocode/article/details/104310781

消息重试分为两种：Producer发送消息的重试和 Consumer消息消费的重试。

一、Producer端重试

Producer端重试是指: Producer往MQ上发消息没有发送成功，比如网络原因导致生产者发送消息到MQ失败。

部分源码解析：

 1    /**
 2      * 说明 抽取部分代码
 3      */
 4     private SendResult sendDefaultImpl(Message msg, final CommunicationMode communicationMode, final SendCallback sendCallback, final long timeout) {
 5         
 6         //1、获取当前时间
 7         long beginTimestampFirst = System.currentTimeMillis();
 8         long beginTimestampPrev ;
 9         //2、去服务器看下有没有主题消息
10         TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
11         if (topicPublishInfo != null && topicPublishInfo.ok()) {
12             boolean callTimeout = false;
13             //3、通过这里可以很明显看出 如果不是同步发送消息 那么消息重试只有1次
14             int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1;
15             //4、根据设置的重试次数，循环再去获取服务器主题消息
16             for (times = 0; times < timesTotal; times++) {
17                 MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);
18                 beginTimestampPrev = System.currentTimeMillis();
19                 long costTime = beginTimestampPrev - beginTimestampFirst;
20                 //5、前后时间对比 如果前后时间差 大于 设置的等待时间 那么直接跳出for循环了 这就说明连接超时是不进行多次连接重试的
21                 if (timeout < costTime) {
22                     callTimeout = true;
23                     break;
24 
25                 }
26                 //6、如果超时直接报错
27                 if (callTimeout) {
28                     throw new RemotingTooMuchRequestException("sendDefaultImpl call timeout");
29                 }
30         }
31     }

通过这段源码很明显可以看出以下几点

如果是异步发送 那么重试次数只有1次
对于同步而言，超时异常也是不会再去重试。
如果发生重试是在一个for 循环里去重试，所以它是立即重试而不是隔一段时间去重试。

实践出真知！！！

二、 Consumer端重试

消费端比较有意思，而且在实际开发过程中，我们也更应该考虑的是消费端的重试。

消费者端的失败主要分为2种情况，Exception 和 Timeout。

1、Exception

 1 @Slf4j
 2 @Component
 3 public class Consumer {
 4     /**
 5      * 消费者实体对象
 6      */
 7     private DefaultMQPushConsumer consumer;
 8     /**
 9      * 消费者组
10      */
11     public static final String CONSUMER_GROUP = "test_consumer";
12     /**
13      * 通过构造函数 实例化对象
14      */
15     public Consumer() throws MQClientException {
16         consumer = new DefaultMQPushConsumer(CONSUMER_GROUP);
17         consumer.setNamesrvAddr("47.99.203.55:9876;47.99.203.55:9877");
18         //订阅topic和 tags（ * 代表所有标签)下信息
19         consumer.subscribe("topic_family", "*");
20         //注册消费的监听 并在此监听中消费信息，并返回消费的状态信息
21         consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
22             //1、获取消息
23             Message msg = msgs.get(0);
24             try {
25                 //2、消费者获取消息
26                 String body = new String(msg.getBody(), "utf-8");
27                 //3、获取重试次数
28                 int count = ((MessageExt) msg).getReconsumeTimes();
29                 log.info("当前消费重试次数为 = {}", count);
30                 //4、这里设置重试大于3次 那么通过保存数据库 人工来兜底
31                 if (count >= 2) {
32                     log.info("该消息已经重试3次,保存数据库。topic={},keys={},msg={}", msg.getTopic(), msg.getKeys(), body);
33                     return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
34                 }
35                 //直接抛出异常
36                 throw new Exception("=======这里出错了============");
37                 //return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
38             } catch (Exception e) {
39                 e.printStackTrace();
40                 return ConsumeConcurrentlyStatus.RECONSUME_LATER;
41             }
42         });
43         //启动监听
44         consumer.start();
45     }
46 }

这里的代码意思很明显: 主动抛出一个异常，然后如果超过3次，那么就不继续重试下去，而是将该条记录保存到数据库由人工来兜底。

看下运行结果

注意 消费者和生产者的重试还是有区别的，主要有两点

1、默认重试次数：Product默认是2次，而Consumer默认是16次。

2、重试时间间隔：Product是立刻重试，而Consumer是有一定时间间隔的。它照1S,5S,10S,30S,1M,2M····2H进行重试。
3、Product在异步情况重试失效，而对于Consumer在广播情况下重试失效。

2、Timeout

说明 这里的超时异常并非真正意义上的超时，它指的是指获取消息后，因为某种原因没有给RocketMQ返回消费的状态，即没有return ConsumeConcurrentlyStatus.CONSUME_SUCCESS 或 return ConsumeConcurrentlyStatus.RECONSUME_LATER。

那么 RocketMQ会认为该消息没有发送，会一直发送。因为它会认为该消息根本就没有发送给消费者,所以肯定没消费。

做这个测试很简单。

 1         //1、消费者获得消息
 2         String body = new String(msg.getBody(), "utf-8");
 3         //2、获取重试次数
 4         int count = ((MessageExt) msg).getReconsumeTimes();
 5         log.info("当前消费重试次数为 = {}", count);
 6         //3、这里睡眠60秒
 7         Thread.sleep(60000);
 8        log.info("休眠60秒 看还能不能走到这里。topic={},keys={},msg={}", msg.getTopic(), msg.getKeys(), body);
 9         //返回成功
10         return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

当获得当前消费重试次数为 = 0 后 , 关掉该进程。再重新启动该进程，那么依然能够获取该条消息

consumer消费者当前消费重试次数为 = 0
休眠60秒看还能不能走到这里。topic=topic_family,keys=1a2b3c4d5f,msg=小小今年3岁

其他理解

首先，我们需要明确，只有当消费模式为 MessageModel.CLUSTERING(集群模式) 时，Broker 才会自动进行重试，对于广播消息是不会重试的。

集群消费模式下，当消息消费失败，RocketMQ 会通过消息重试机制重新投递消息，努力使该消息消费成功。

当消费者消费该重试消息后，需要返回结果给 broker，告知 broker 消费成功（ConsumeConcurrentlyStatus.CONSUME*SUCCESS）或者需要重新消费（ConsumeConcurrentlyStatus.RECONSUME*LATER）

这里有个问题，如果消费者业务本身故障导致某条消息一直无法消费成功，难道要一直重试下去吗？

答案是显而易见的，并不会一直重试。

事实上，对于一直无法消费成功的消息，RocketMQ 会在达到最大重试次数之后，将该消息投递至死信队列。然后我们需要关注死信队列，并对该死信消息业务做人工的补偿操作。

那如何返回消息消费失败呢？

RocketMQ 规定，以下三种情况统一按照消费失败处理并会发起重试。

业务消费方返回 ConsumeConcurrentlyStatus.RECONSUME_LATER
业务消费方返回null
业务消费方主动/被动抛出异常

前两种情况较容易理解，当返回 ConsumeConcurrentlyStatus.RECONSUME_LATER或者 null时，broker 会知道消费失败，后续就会发起消息重试，重新投递该消息。

注意对于抛出异常的情况，只要我们在业务逻辑中显式抛出异常或者非显式抛出异常，broker 也会重新投递消息，如果业务对异常做了捕获，那么该消息将不会发起重试。因此对于需要重试的业务，消费方在捕获异常的时候要注意返回 ConsumeConcurrentlyStatus.RECONSUME*LATER 或 null 并输出异常日志，打印当前重试次数。（推荐返回ConsumeConcurrentlyStatus.RECONSUME*LATER）

死信的业务处理方式

默认的处理机制中，如果我们只对消息做重复消费，达到最大重试次数之后消息就进入死信队列了。

我们也可以根据业务的需要，定义消费的最大重试次数，每次消费的时候判断当前消费次数是否等于最大重试次数的阈值。

如：重试三次就认为当前业务存在异常，继续重试下去也没有意义了，那么我们就可以将当前的这条消息进行提交，返回 broker 状态ConsumeConcurrentlyStatus.CONSUME_SUCCES，让消息不再重发，同时将该消息存入我们业务自定义的死信消息表，将业务参数入库，相关的运营通过查询死信表来进行对应的业务补偿操作。

RocketMQ 的处理方式为将达到最大重试次数（16 次）的消息标记为死信消息，将该死信消息投递到 DLQ 死信队列中，业务需要进行人工干预。实现的逻辑在 SendMessageProcessor的 consumerSendMsgBack方法中，大致思路为首先判断重试次数是否超过 16 或者消息发送延时级别是否小于 0，如果已经超过 16 或者发送延时级别小于 0，则将消息设置为新的死信。死信 topic 为：%DLQ%+consumerGroup

我们接着看一下死信的源码实现机制。

 1 private RemotingCommand consumerSendMsgBack(final ChannelHandlerContext ctx, final RemotingCommand request)
 2             throws RemotingCommandException {
 3             final RemotingCommand response = RemotingCommand.createResponseCommand(null);
 4             final ConsumerSendMsgBackRequestHeader requestHeader =
 5                 (ConsumerSendMsgBackRequestHeader)request.decodeCommandCustomHeader(ConsumerSendMsgBackRequestHeader.class);
 6 
 7             ......
 8 
 9             // 0.首先判断重试次数是否大于等于 16，或者消息延迟级别是否小于 0
10             if (msgExt.getReconsumeTimes() >= maxReconsumeTimes
11                 || delayLevel < 0) {
12                 // 1. 如果满足判断条件，设置死信队列 topic= %DLQ%+consumerGroup
13                 newTopic = MixAll.getDLQTopic(requestHeader.getGroup());
14                 queueIdInt = Math.abs(this.random.nextInt() % 99999999) % DLQ_NUMS_PER_GROUP;
15 
16                 topicConfig = this.brokerController.getTopicConfigManager().createTopicInSendMessageBackMethod(newTopic,
17                     DLQ_NUMS_PER_GROUP,
18                     PermName.PERM_WRITE, 0
19                 );
20                 if (null == topicConfig) {
21                     response.setCode(ResponseCode.SYSTEM_ERROR);
22                     response.setRemark("topic[" + newTopic + "] not exist");
23                     return response;
24                 }
25             } else {
26                 // 如果延迟级别为 0，则设置下一次延迟级别为 3+当前重试消费次数，达到时间衰减效果
27                 if (0 == delayLevel) {
28                     delayLevel = 3 + msgExt.getReconsumeTimes();
29                 }
30 
31                 msgExt.setDelayTimeLevel(delayLevel);
32             }
33 
34             MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
35             msgInner.setTopic(newTopic);
36             msgInner.setBody(msgExt.getBody());
37             msgInner.setFlag(msgExt.getFlag());
38             MessageAccessor.setProperties(msgInner, msgExt.getProperties());
39             msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgExt.getProperties()));
40             msgInner.setTagsCode(MessageExtBrokerInner.tagsString2tagsCode(null, msgExt.getTags()));
41 
42             msgInner.setQueueId(queueIdInt);
43             msgInner.setSysFlag(msgExt.getSysFlag());
44             msgInner.setBornTimestamp(msgExt.getBornTimestamp());
45             msgInner.setBornHost(msgExt.getBornHost());
46             msgInner.setStoreHost(this.getStoreHost());
47             msgInner.setReconsumeTimes(msgExt.getReconsumeTimes() + 1);
48 
49             String originMsgId = MessageAccessor.getOriginMessageId(msgExt);
50             MessageAccessor.setOriginMessageId(msgInner, UtilAll.isBlank(originMsgId) ? msgExt.getMsgId() : originMsgId);
51 
52             // 3.死信消息投递到死信队列中并落盘
53             PutMessageResult putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner);
54             ......
55             return response;
56         }

我们总结一下死信的处理逻辑：

首先判断消息当前重试次数是否大于等于 16，或者消息延迟级别是否小于 0
只要满足上述的任意一个条件，设置新的 topic（死信 topic）为：%DLQ%+consumerGroup
进行前置属性的添加
将死信消息投递到上述步骤 2 建立的死信 topic 对应的死信队列中并落盘，使消息持久化。