大家好，我是大堯。

1. 為什么你們公司選擇RabbitMQ作為消息中間件

在消息隊列選型時，我們調(diào)研了市場上比較常用ActiveMQ，RabbitMQ，RocketMQ，Kafka。

1. RabbitMQ相對成熟穩(wěn)定，這是我們選擇它最主要的原因。
2. 社區(qū)比較活躍，有完善的資料可以參考。
3. Rabbitmq的吞吐量可以達到萬級，完全滿足我們系統(tǒng)的要求。
4. RabbitMQ是Erlang語言開發(fā)的，性能比較好。
5. 有完善的可視化界面，方便查看。

2. 消息隊列的優(yōu)點和缺點有哪些

優(yōu)點有：

• 異步處理 - 相比于傳統(tǒng)的串行、并行方式，提高了系統(tǒng)吞吐量。
• 應(yīng)用解耦 - 系統(tǒng)間通過消息通信，不用關(guān)心其他系統(tǒng)的處理。
• 流量削鋒 - 可以通過消息隊列長度控制請求量；可以緩解短時間內(nèi)的高并發(fā)請求。

缺點有：

• 系統(tǒng)可用性降低
• 系統(tǒng)復雜度提高

3. RabbitMQ常用的工作模式有哪些

3.1 簡單模型

• p：生成者
• C：消費者
• 紅色部分：quene，消息隊列

3.2 工作模型

這種模式下一條消息只能由一個消費者進行消費，默認情況下，每個消費者是輪詢消費的。

• p：生成者
• C1、C2：消費者
• 紅色部分：quene，消息隊列

3.3 發(fā)布訂閱模型（fanout）

這種模型中生產(chǎn)者發(fā)送的消息所有消費者都可以消費。

• p：生成者
• X：交換機
• C1、C2：消費者
• 紅色部分：quene，消息隊列

3.4 路由模型（routing）

這種模型消費者發(fā)送的消息，不同類型的消息可以由不同的消費者去消費。

• p：生成者
• X：交換機，接收到生產(chǎn)者的消息后將消息投遞給與routing key完全匹配的隊列
• C1、C2：消費者
• 紅色部分：quene，消息隊列

3.5 主題模型（topic）

這種模型和direct模型一樣，都是可以根據(jù)routing key將消息路由到不同的隊列，只不過這種模型可以讓隊列綁定routing key 的時候使用通配符。這種類型的routing key都是由一個或多個單詞組成，多個單詞之間用.分割。

通配符介紹：

*：只匹配一個單詞

#：匹配一個或多個單詞

4. 如何保證消息不丟失（如何保證消息的可靠性）

一條消息從生產(chǎn)到消費經(jīng)歷了三個階段，分別是生產(chǎn)者，MQ和消費者，對于RabbitMQ來說，消息的傳遞還涉及到交換機。因此RabbitMQ出現(xiàn)消息丟失的情況有四個

分別是

1. 消息生產(chǎn)者沒有成功將消息發(fā)送到MQ導致消息丟失
2. 交換機未路由到消息隊列導致消息丟失
3. 消息在MQ中時，MQ發(fā)生宕機導致消息丟失
4. 消費者消費消息時出現(xiàn)異常導致消息丟失

針對上面提到的四種情況，分別進行處理

1. amqp協(xié)議提供了事務(wù)機制，在投遞消息時開啟事務(wù)，如果消息投遞失敗，則回滾事務(wù)，很少有人去使用事務(wù)。除了事務(wù)之外，RabbitMQ還提供了生產(chǎn)者確認機制（publisher confirm）。生產(chǎn)者將信道設(shè)置成confirm（確認）模式，一旦信道進入confirm模式，所有在該信道上面發(fā)布的消息都會被指派一個唯一的ID（從1開始），一旦消息被投遞到所有匹配的隊列之后，RabbitMQ就會發(fā)送一個確認（Basic.Ack）給生產(chǎn)者（包含消息的唯一ID），這就使得生產(chǎn)者知曉消息已經(jīng)正確到達了目的地了。

# 開啟生產(chǎn)者確認機制，
# 注意這里確認的是是否到達交換機
spring.rabbitmq.publisher-confirm-type=correlated
@RestController
public?class?Producer?{

????@Autowired
????private?RabbitTemplate?rabbitTemplate;

????@GetMapping("send")
????public?void?sendMessage(){
????????/**
?????????*?生產(chǎn)者確認消息
?????????*/
????????rabbitTemplate.setConfirmCallback(new?RabbitTemplate.ConfirmCallback()?{
????????????@Override
????????????public?void?confirm(CorrelationData?correlationData,?boolean?ack,?String?cause)?{
????????????????System.out.println(correlationData);
????????????????System.out.println(ack);
????????????????System.out.println(cause);
????????????}
????????});
????????rabbitTemplate.convertAndSend("s","error","這是一條錯誤日志！??！");
????}
}


2. 消息從交換機未能匹配到隊列時將此條消息返回給生產(chǎn)者

spring.rabbitmq.publisher-returns=true
@RestController
public?class?Producer?{

????@Autowired
????private?RabbitTemplate?rabbitTemplate;

????@GetMapping("send")
????public?void?sendMessage(){
????????/**
?????????*?消息未達隊列時返回該條消息
?????????*/
????????rabbitTemplate.setReturnsCallback(new?RabbitTemplate.ReturnsCallback()?{
????????????@Override
????????????public?void?returnedMessage(ReturnedMessage?returnedMessage)?{
????????????????System.out.println(returnedMessage);
????????????}
????????});
????????rabbitTemplate.convertAndSend("s","error","這是一條錯誤日志?。?！");
????}
}


3. 消息在交換機或隊列中發(fā)生丟失，我們只需要將交換機和隊列進行持久化。

/**
??*?定義一個持久化的topic交換機
??*?durable?持久化
??*?@return
?*/
@Bean
public?Exchange?exchangeJavatrip(){
????return?ExchangeBuilder.topicExchange(EXCHANGE).durable(true).build();
}

/**
?*?定義一個持久化的隊列
?*?durable?持久化
?*?@return
?*/
@Bean
public?Queue?queueJavatrip(){
????return?QueueBuilder.durable(QUEUE).build();
}


4. 消費者開啟手動簽收模式，消費完成后進行ack確認。

spring.rabbitmq.listener.simple.acknowledge-mode=manual
@RabbitListener(queues?=?MqConfig.QUEUE)
public?void?receive(String?body,?Message?message,?Channel?channel)?throws?Exception{
????long?deliveryTag?=?message.getMessageProperties().getDeliveryTag();
????System.out.println(deliveryTag);
????//?系統(tǒng)業(yè)務(wù)邏輯判斷是否簽收
????if(deliveryTag?%?2?==?0){
????????channel.basicAck(deliveryTag,false);
????}else{
????????//?第二個參數(shù)是否批量確認，第三個參數(shù)是否重新回隊列
????????channel.basicNack(deliveryTag,false,true);
????}
}


5. 如何保證消息不重復消費（如何保證消息的冪等性）

消息重復的原因有兩個：

1. 生產(chǎn)時消息重復
   由于生產(chǎn)者發(fā)送消息給MQ，在MQ確認的時候出現(xiàn)了網(wǎng)絡(luò)波動，生產(chǎn)者沒有收到確認，實際上MQ已經(jīng)接收到了消息。這時候生產(chǎn)者就會重新發(fā)送一遍這條消息。

2. 消費時消息重復。
   消費者消費成功后，在給MQ確認的時候出現(xiàn)了網(wǎng)絡(luò)波動，MQ沒有接收到確認，為了保證消息被消費，MQ就會繼續(xù)給消費者投遞之前的消息。這時候消費者就接收到了兩條一樣的消息。

由于消息重復是網(wǎng)絡(luò)波動等原因造成的，無法避免，我們能做的的就是保證消息的冪等性，以防業(yè)務(wù)重復處理。具體處理方案為：

讓每個消息攜帶一個全局的唯一ID，即可保證消息的冪等性，具體消費過程為：

1. 消費者獲取到消息后先根據(jù)id去查詢redis/db是否存在該消息。
2. 如果不存在，則正常消費，消費完畢后寫入redis/db。
3. 如果存在，則證明消息被消費過，直接丟棄。

@RabbitListener(queues?=?MqConfig.QUEUE)
public?void?receive(Message?message,?Channel?channel){

????String?messageId?=?message.getMessageProperties().getMessageId();
????String?body?=?new?String(message.getBody());
????String?redisId?=?redisTemplate.opsForValue().get(messageId) "";
????//?如果redis中存有當前消息的消息id
????//?則證明消費過
????if(messageId.equals(redisId)){
????????return;
????}
????redisTemplate.opsForValue().set(messageId,?UUID.randomUUID());
}


6. 消息大量堆積應(yīng)該怎么處理

消息堆積的原因有兩個

1. 網(wǎng)絡(luò)故障，消費者無法正常消費
2. 消費方消費后未進行ack確認

解決方案如下：

1. 檢查并修復消費者故障，使其正常消費
2. 編寫臨時程序?qū)⒍逊e的消息發(fā)送到容量更大的MQ集群，增加消費者快速消費
3. 堆積消息消費完畢后，停止臨時程序，恢復正常消費

7. 死信是什么？死信如何處理

當一條消息在隊列中出現(xiàn)以下三種情況的時候，該消息就會變成一條死信。

• 消息被拒絕(basic.reject / basic.nack)，并且requeue = false
• 消息TTL過期
• 隊列達到最大長度

當消息在一個隊列中變成一個死信之后，如果配置了死信隊列，它將被重新publish到死信交換機，死信交換機將死信投遞到一個隊列上，這個隊列就是死信隊列。

一條消息成為死信后，一般會通過死信隊列進行存庫，然后定時將庫中的死信進行重新投遞到消息隊列上。

8. 如果我有一筆訂單，30分鐘未支付則關(guān)閉訂單，使用RabbitMQ如何來實現(xiàn)

RabbitMQ可以使用死信隊列來實現(xiàn)延時消費，用戶下單之后，將訂單信息投遞到消息隊列中，并且設(shè)置消息過期時常為30分鐘。如果用戶支付則正常關(guān)閉訂單，如果用戶未支付，消息達到過期時間，消息會進入死信交換，由消費者進行消費死信隊列來關(guān)閉訂單。

9. RabbitMQ如何保證高可用

RabbitMQ有兩種集群模式，分別是普通集群和鏡像集群，普通模式無法保證RabbitMQ的高可用。

普通集群

假如有三個節(jié)點，rabbitmq1、rabbitmq2、rabbitmq3，消息實際上只存在于其中一個節(jié)點，三個節(jié)點僅有相同的元數(shù)據(jù)，即隊列的結(jié)構(gòu)，當消息進入rabbitmq2節(jié)點的queue后，consumer從rabbitmq1的節(jié)點進行消費，rabbitmq1和rabbitmq2會進行臨時通信，從rabbitmq2中獲取消息然后返回給consumer。

這種模式存在以下兩個問題：

1. 當rabbitmq2宕機后，消息無法正常消費，沒有做到真正的高可用

2. 實際數(shù)據(jù)還是在單個實例上，存在瓶頸問題

鏡像集群

假如有三個節(jié)點，rabbitmq1、rabbitmq2、rabbitmq3，每個實例之間都可以相互通信，每次生產(chǎn)者寫消息到queue的時候，每個rabbitmq節(jié)點上都有queue的消息數(shù)據(jù)和元數(shù)據(jù)。這種模式適用于可靠性要求較高的場景。